How to create a refreshable table using in-memory data in Flink for joins?

Question

I have a Flink application that I rely on Table API. I do have a Kafka topic that I create a table. Then, we maintain an S3 object for list of IP addressed and some metadata information.

We also want to create a table on this S3 object. S3 object path is static and does not change, but I can override the S3 object and I want to refresh this table with the new data.

Basically, I have a collection in-memory read from the S3 object. How can I create a table and do join on the Kafka table most efficiently? The table should be refreshed when there is an update in S3 object.

David Anderson · Accepted Answer · 2020-11-12 18:41:33Z

1

If you create a Table that is backed by the S3 object, using the FileSystem SQL Connector, it might do what you are looking for. Note, however, that file system sources are not fully developed, and you may run into limitations that affect your use case.

You could instead use StreamExecutionEnvironment#readFile (docs), and convert the DataStream that it produces into a Table. Note that if you read a file with readFile while using FileProcessingMode.PROCESS_CONTINUOUSLY mode, and then modify the file, the entire file will be re-ingested.

edited Nov 12, 2020 at 18:41

answered Nov 12, 2020 at 9:53

David Anderson

44.3k4 gold badges41 silver badges73 bronze badges

Sign up to request clarification or add additional context in comments.

7 Comments

lalala Over a year ago

Isn't the connector expecting a directory(prefix)instead of an object path?

David Anderson Over a year ago

I'm not sure; I know the filesystem SQL sink works that way. I've expanded my answer to include an alternative.

lalala Over a year ago

Thank you, I guess PROCESS_CONTINUOUSLY would work. If I don't misunderstand, it will monitor the file and will re-ingest. I hope it will drop the existing content when new file comes, and do this atomically.

David Anderson Over a year ago

No, it won't happen atomically.

lalala Over a year ago

Do you think I can use the concept of Temporal Tables. For example, something similar to Hive Temporal Tabel(ci.apache.org/projects/flink/flink-docs-stable/dev/table/hive/…). My data volume is relatively small, I just need a proper abstraction that I can refresh my metadata and provide an SQL interface for joins

|

Collectives™ on Stack Overflow

How to create a refreshable table using in-memory data in Flink for joins?

1 Answer 1

7 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

7 Comments

Your Answer

Sign up or log in

Post as a guest

Related