JoinStreams.java

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

    // assumes PLAINTEXT authentication
    KafkaSource<Product> productSource = KafkaSource.<Product>builder()
            .setBootstrapServers(prop.getProperty("BOOTSTRAP_SERVERS"))
            .setTopics(prop.getProperty("PRODUCTS_TOPIC"))
            .setGroupId("flink_join_demo")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(new ProductDeserializationSchema())
            .build();

    DataStream<Product> productsStream = env.fromSource(
            productSource, WatermarkStrategy.noWatermarks(), "Kafka Products Source");

    tableEnv.createTemporaryView("products", productsStream);

    KafkaSource<Purchase> purchasesSource = KafkaSource.<Purchase>builder()
            .setBootstrapServers(prop.getProperty("BOOTSTRAP_SERVERS"))
            .setTopics(prop.getProperty("PURCHASES_TOPIC"))
            .setGroupId("flink_join_demo")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(new PurchaseDeserializationSchema())
            .build();

    DataStream<Purchase> purchasesStream = env.fromSource(
            purchasesSource, WatermarkStrategy.noWatermarks(), "Kafka Purchases Source");

    tableEnv.createTemporaryView("purchases", purchasesStream);

    Table result =
            tableEnv.sqlQuery(
                    "SELECT " +
                            "purchases.transactionTime, " +
                            "TO_TIMESTAMP(purchases.transactionTime), " +
                            "purchases.transactionId, " +
                            "purchases.productId, " +
                            "products.category, " +
                            "products.item, " +
                            "products.size, " +
                            "products.cogs, " +
                            "products.price, " +
                            "products.containsFruit, " +
                            "products.containsVeggies, " +
                            "products.containsNuts, " +
                            "products.containsCaffeine, " +
                            "purchases.price, " +
                            "purchases.quantity, " +
                            "purchases.isMember, " +
                            "purchases.memberDiscount, " +
                            "purchases.addSupplements, " +
                            "purchases.supplementPrice, " +
                            "purchases.totalPurchase " +
                            "FROM " +
                            "products " +
                            "JOIN purchases " +
                            "ON products.productId = purchases.productId"
            );

    DataStream<PurchaseEnriched> purchasesEnrichedTable = tableEnv.toDataStream(result,
            PurchaseEnriched.class);

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

    purchasesEnrichedTable.sinkTo(sink);

    env.execute("Flink Streaming Join Demo");
}