CODEWITHSUNDEEP
apache-flinkflink-streaming
clay
clay

Reputation: 20460

Flink: DataStream left join Table. Super simple

        DataStream<String> sourceStream = streamEnv.fromElements("key_a", "key_b", "key_c", "key_d");

        Table lookupTable = tableEnv.fromValues(
                DataTypes.ROW(
                        DataTypes.FIELD("my_key", DataTypes.STRING()),
                        DataTypes.FIELD("my_value", DataTypes.STRING())
                ),
                Expressions.row("key_a", "value_a"),
                Expressions.row("key_b", "value_b")
        );

I'd like to left join the stream to the table.

This is obviously a simplified demo scenario. I'd like to understand how to use the Flink API to achieve this with a toy data set before using a larger production data set.

The docs on Table joins shows how to join two tables and get another table back, which is not what I want:

https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/tableApi.html#joins

The docs on DataStream joins shows to join two streams on a time window, which is also not what I want:

https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/stream/operators/joining.html

Upvotes: 0

Views: 1264

Answers (1)

David Anderson
David Anderson

Reputation: 43707

I believe this is what you are looking for. This example converts the sourceStream to a dynamic table, joins it with the lookup table, and then converts the resulting dynamic table back to a stream for printing.

You could, instead, do further processing on the resultStream using the DataStream API.

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

public class JoinExample {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStream<String> sourceStream = env.fromElements("key_a", "key_b", "key_c", "key_d");
        Table streamTable = tableEnv.fromDataStream(sourceStream, $("stream_key"));

        Table lookupTable = tableEnv.fromValues(
                DataTypes.ROW(
                        DataTypes.FIELD("lookup_key", DataTypes.STRING()),
                        DataTypes.FIELD("lookup_value", DataTypes.STRING())
                ),
                Expressions.row("key_a", "value_a"),
                Expressions.row("key_b", "value_b")
        );

        Table resultTable = streamTable
                .join(lookupTable).where($("stream_key").isEqual($("lookup_key")))
                .select($("stream_key"), $("lookup_value"));

        DataStream<Row> resultStream = tableEnv.toAppendStream(resultTable, Row.class);

        resultStream.print();

        env.execute();
    }
}

The output is

key_b,value_b
key_a,value_a

Upvotes: 2

Related Questions