Table API & PyFlink — one pipeline, two languages

4 h
requires FLK-04

From your world. a dbt model expressed as a relational pipeline, written once and runnable from Java or Python (like dbt SQL vs dbt Python models)

← All exercises

Per-currency GMV in Java Table API and PyFlink

Write the same pipeline twice. Write it once in the Java Table API and once in PyFlink. Then prove that both compute the same answer. The pipeline is a per-currency gross-merchandise-value (GMV): group orders by currency and sum the amount. It is the streaming analogue of a dbt SQL model and the equivalent dbt Python model.

Java — `de.flk.GmvTable`

public static Map<String, Long> run(List<String[]> orders) throws Exception

orders rows are {ccy, amtMinor} (amount in minor units as a decimal string).
Build a TableEnvironment in batch mode and make a table from the rows. Group by ccy and sum amtMinor. Return ccy -> total as a Map<String,Long>. You may use the Table API (groupBy(...).select(...)) or executeSql("SELECT ccy, SUM(amt) ... GROUP BY ccy"). The grader checks the numbers, not the API.

PyFlink — `job.py`

Read the same orders from the environment variable ORDERS (format: ccy,amt;ccy,amt;...). Run the same per-currency GMV with the Table API of PyFlink (or execute_sql). Print exactly one line:

DE_RESULT={"EUR": 1250000, "USD": 500000}

That is, use print("DE_RESULT=" + json.dumps(totals, sort_keys=True)). Here totals maps each currency to the integer sum of amtMinor.

Grading

The grader runs the Java method (through the embedded Flink runner) and the PyFlink script (in its own venv) over the same orders. It compares each result to one independent Python oracle (SUM(amt) GROUP BY ccy). Both must match. The PyFlink test skips cleanly if the PyFlink environment is not installed. The Java Table API test always grades.

Why the starter fails

The Java method returns an empty map and job.py prints nothing useful. Implement both, so each reproduces the per-currency totals.

Starter code

job.py

import json

# Starter: not implemented. Prints an empty result so the PyFlink test fails (G1).
#
# TODO: read ORDERS from the environment, build a PyFlink TableEnvironment in
# batch mode, run a per-currency SUM(amt) GROUP BY ccy, and print
# "DE_RESULT=" + json.dumps(totals, sort_keys=True).


def main():
    totals = {}
    print("DE_RESULT=" + json.dumps(totals, sort_keys=True))


if __name__ == "__main__":
    main()

src/main/java/de/flk/GmvTable.java

package de.flk;

import java.util.*;

/**
 * Starter: not implemented. Returns an empty map so the Java test fails (G1).
 *
 * Build a TableEnvironment in batch mode, group the orders by ccy, sum amt, and
 * return ccy -> total.
 */
public class GmvTable {
    public static Map<String, Long> run(List<String[]> orders) throws Exception {
        // TODO: Table API groupBy(ccy).select(ccy, amt.sum()) or executeSql.
        return new LinkedHashMap<>();
    }
}

Hints

Hint 1

Java: TableEnvironment.create(EnvironmentSettings.inBatchMode()); build a table with fromValues, then groupBy($("ccy")).select($("ccy"), $("amt").sum()) OR just executeSql a GROUP BY. Collect the Rows into a Map<String,Long>.

Hint 2

PyFlink: TableEnvironment.create(EnvironmentSettings.in_batch_mode()); from_elements or execute_sql a GROUP BY; iterate result.collect() and print 'DE_RESULT=' + json.dumps(totals, sort_keys=True).

Hint 3

Both must return the SAME per-currency totals (SUM of amtMinor as a long). The grader compares each language's output against one Python oracle; the APIs may differ, the numbers may not.

Run it locally

uv run de start FLK-08/01-gmv-two-languages
# edit workspace/FLK-08/01-gmv-two-languages/
uv run de test