https://flink.apache.org/ logo
Join Slack
Powered by
Can someone confirm that, if I create N tables thr...
# troubleshooting
v
Can someone confirm that, if I create N tables through the Table API, all using the Kafka source connector and reading different topics, then N kafka consumers will be created? If that's the case, is there any way I can optimize it so it only creates one Kafka consumer?
m
When you’re creating a table, it doesn’t start reading anything until you’re creating a select/insert query that uses those tables
And those you can optimize by using a statement set
v
I understand that part. I'm confused if there will be one Kafka consumer reading all the topics needed and then "splitting". Or one table translates to a consumer. For example, if I have:
Copy code
CREATE TABLE table1 (
  ...
) WITH (
  'connector' = 'kafka',
  'topic' = 'topic1',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

CREATE TABLE table2 (
  ...
) WITH (
  'connector' = 'kafka',
  'topic' = 'topic2',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);
and I then submit a job that uses both tables to compute something, then will I end up with one consumer?
With the Datastreams API, as far as I know, I can create a single Kafka source that reads N topics, and then split the messages received using OutputTags. It's a bit more work. I was wondering if the Table API sort of does that already.
s
With the Datastreams API, as far as I know, I can create a single Kafka source that reads N topics, and then split the messages received using OutputTags. It’s a bit more work. I was wondering if the Table API sort of does that already.
No, you’ll end up with N consumers. You can only optimize it by resorting to the DataStream API, splitting messages and converting individual streams to tables.
✅ 1
v
Right... we have a use case where we will be reading from 500 different topics; in this case you think it's better to just use the Datastreams API then?
We ran into some OOM issues and I think that's why 😕
s
Yep, I think so! Especially if some of them have similar schemas.
d
Why doesn't it work to use topic discovery with the SQL Kafka source, and then define some views on that uber table to split it down to the individual tables? Put all that in a statement set, of course.
v
@David Anderson all topics have different schemas unfortunately
unless there's a way to accommodate every schema in one table somehow in the Table API? i'm not aware of anything like it
i was thinking of creating one table with one 
variant
column (not sure if something like this exists in Flink) and then create views, filtering by name
something like:
Copy code
CREATE TABLE alltopics (
  name STRING,
  data VARIANT
) WITH (
  'connector' = 'kafka',
  'topic.list' = 'topic1,topic2',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

CREATE VIEW table1 AS SELECT data->field1, data->field2 FROM alltopics WHERE name = 'table1';
CREATE VIEW table2 AS SELECT data->field1, data->field2 FROM alltopics WHERE name = 'table2';
unfortunately I don't think flink has anything like jsonb, maybe i could convert the json to a map somehow
there is 
JSON_VALUE
though.
d
You could, perhaps, define the column as RAW, and then use one or more UDFs to deserialize the content in a set of views.
Open in Slack
PreviousNext
Powered by