RunningTotals.java

public static void flinkKafkaPipeline(Properties prop) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // assumes PLAINTEXT authentication
    KafkaSource<Purchase> source = KafkaSource.<Purchase>builder()
            .setBootstrapServers(prop.getProperty("BOOTSTRAP_SERVERS"))
            .setTopics(prop.getProperty("PURCHASES_TOPIC"))
            .setGroupId("flink_reduce_demo")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(new PurchaseDeserializationSchema())
            .build();

    DataStream<Purchase> purchases = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

    DataStream<RunningTotal> runningTotals = purchases
            .flatMap((FlatMapFunction<Purchase, RunningTotal>) (purchase, out) -> out.collect(
                    new RunningTotal(
                            purchase.getTransactionTime(),
                            purchase.getProductId(),
                            1,
                            purchase.getQuantity(),
                            purchase.getTotalPurchase()
                    ))
            ).returns(RunningTotal.class)
            .keyBy(RunningTotal::getProductId)
            .reduce((runningTotal1, runningTotal2) -> {
                runningTotal2.setTransactions(runningTotal1.getTransactions() + runningTotal2.getTransactions());
                runningTotal2.setQuantities(runningTotal1.getQuantities() + runningTotal2.getQuantities());
                runningTotal2.setSales(runningTotal1.getSales().add(runningTotal2.getSales()));
                return runningTotal2;
            });

    KafkaSink<RunningTotal> sink = KafkaSink.<RunningTotal>builder()
            .setBootstrapServers(prop.getProperty("BOOTSTRAP_SERVERS"))
            .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                    .setTopic(prop.getProperty("RUNNING_TOTALS_TOPIC"))
                    .setValueSerializationSchema(new RunningTotalSerializationSchema())
                    .build()
            ).setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
            .build();

    runningTotals.sinkTo(sink);

    env.execute("Flink Running Totals Demo");
}