Hello. I am currently working on a data streaming project and our team decided to use Table API to do data processing on that are sending through the Kafka Producer. We are sending thousands of messages on two different topics. Every single message contains a large string (2^14 characters), and the message is being serialized within JSON format. I used the Table API to run query over two tables (J*OIN SQL*) and pass results to a data processing function when they are ready: 1) Is Table API able to handle queries on data that contain large amount of strings? 2) If so, what options are there to make processes faster? 3) I use Pyflink and it does not seem to be fast enough. Is Table API might be faster through Java or Scala? If so, which one is fastest at this point? Thank you very much for any suggestions. Here this is sample of my code:

#### On Producer side:
    for raw_id in tqdm.tqdm(range(0, Nt)):
        data_array = fid.read(chunk_size)
        b64_string = str(base64.b64encode(data_array), 'utf-8')
        empad_json = create_sample_empad_json(raw_id, b64_string, nFileLen, chunk_size)
        # print(b64_string)
        producer.send(topic, empad_json)
        producer.flush()

def create_sample_empad_json(raw_id, raw_data, file_size, chunk_size):
    return {'raw_id': int(raw_id), 'raw_data': raw_data, 'file_size': file_size, 'chunk_size': chunk_size}

#### On Flink side:
def stream_processing():
    env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(stream_execution_environment=env)
    t_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True)

    t1 = f"""
            CREATE TABLE raw_table(
                raw_id INT,
                raw_data STRING,
                file_size INT,
                chunk_size INT
            ) WITH (
              'connector' = 'kafka',
              'topic' = 'empadraw',
              'properties.bootstrap.servers' = 'kafka:9092',
              'properties.group.id' = 'MY_GRP',
              'scan.startup.mode' = 'latest-offset',
              'format' = 'json'
            )
            """

    t2 = f"""
            CREATE TABLE bkgd_table(
                raw_id INT,
                raw_data STRING,
                file_size INT,
                chunk_size INT
            ) WITH (
              'connector' = 'kafka',
              'topic' = 'empadbkgd',
              'properties.bootstrap.servers' = 'kafka:9092',
              'properties.group.id' = 'MY_GRP',
              'scan.startup.mode' = 'latest-offset',
              'format' = 'json'
            )
            """

    t_env.execute_sql(t1)
    t_env.execute_sql(t2)

    query = "SELECT raw_table.raw_id as raw_id, raw_table.raw_data as raw_enc_val, raw_table.file_size as raw_file_size, raw_table.chunk_size as raw_chunk_size, " \
            "bkgd_table.raw_data as bkgd_enc_val, bkgd_table.file_size as bkgd_file_size, bkgd_table.chunk_size as bkgd_chunk_size " \
            "FROM raw_table JOIN bkgd_table ON raw_table.raw_id = bkgd_table.raw_id"

    table_result = t_env.sql_query(query).execute().collect()
    process_results(table_result)

   def process_results(table_result):
    with table_result as results:
        for result in results:
            chunk_id = int(result[0])
            raw_chunk = str(result[1])
            data_chunk = str(result[2])
            file_size = int(result[3])
            chunk_size = int(result[4])

            process_data(chunk_id, raw_chunk, bkgd_chunk, file_size, chunk_size)

I don't have experience with pyflink, but Im not sure if you should see any performance hits - the same should apply between different Flink APIs. Some things to explore: 1. How large are those topics? (messages / second?) 2. How many partitions each topic has? Does it allow for proper scaling? 3. Do you see any data skew inside your topics? 4. I see you don't specify any watermarks on your tables 5. Both topics are append-only streams or one is used as a changelog stream that keeps state? i.e you use that for data enrichment for example? Since you are running a typical join both tables will be kept in memory. Also events will be buffered until a matching record arrives for any side. For example if you read message 1 from topic A and it has a matching message in topic B that is the number 100.000 for example, the message will be buffered until all the message until that specific one are consumed first and build in state. 6. Also since your messages are quite large, there might be some deserialization hit?

Hi @Giannis Polyzos Many thanks for your response and comments: At the moment I am running everything on Docker on my laptop (Apple M1 32 GB RAM). I will ask those specific questions from technical team and will back to you soon. Also you hit the issue by making the example: “if you read message 1 from topic A and it has a matching … consumed first and build in state.” So, what’s the approach for that issue? Also, yes, we are doing our specific serialization/deserialization but in my example I am using a simple JSON serialization. So, if JSON deserialization is not fast enough what options are there? Thanks.

• Why you collect the data to client? If you want to do something for each record, I guess you could convert the table to datastream[1] and then perform datastream.process/datastream.flatmap. Compared with the current approach, it will process the data in distributed manner. • What are you doing in

process_data

, is it efficient enough? • What’s the current performance? [1] https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/python/datastream/intro_to_datastream_api/#conversion-betwee[…]astream-and-table

@Dian Fu Data generate by microscopes from different institutes and contain numerical data. We do some data analysis over process_data function once we access them. Can you show me a document how to convert table to datastream? I don’t intend to collect data to client but my connectors are Kafka (as I posted my sample code) and I am looking for an approach to connect to access kafka messages and process them while they are producing.

This link [1] shows how to convert Table to DataStream, that’s

to_append_stream

[1] https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/python/datastream/intro_to_datastream_api/#conversion-between-datastream-and-table

@Dian Fu thanks. This is how I converted tables to data streams. I hope I am on the right track:

t1 = f"""
            CREATE TABLE raw_table(
                raw_id INT,
                raw_data STRING,
                file_size INT, 
                chunk_size INT
            ) WITH (
              'connector' = 'kafka',
              'topic' = 'empadraw',
              'properties.bootstrap.servers' = 'kafka:9092',
              'properties.group.id' = 'MY_GRP',
              'scan.startup.mode' = 'latest-offset',
              'format' = 'json'
            )
            """

    t2 = f"""
            CREATE TABLE bkgd_table(
                raw_id INT,
                raw_data STRING,
                file_size INT, 
                chunk_size INT
            ) WITH (
              'connector' = 'kafka',
              'topic' = 'empadbkgd',
              'properties.bootstrap.servers' = 'kafka:9092',
              'properties.group.id' = 'MY_GRP',
              'scan.startup.mode' = 'latest-offset',
              'format' = 'json'
            )
            """

    t_env.execute_sql(t1)
    t_env.execute_sql(t2)

    query = "SELECT raw_table.raw_id as raw_id, raw_table.raw_data as raw_enc_val, raw_table.file_size as raw_file_size, raw_table.chunk_size as raw_chunk_size, " \
            "bkgd_table.raw_data as bkgd_enc_val, bkgd_table.file_size as bkgd_file_size, bkgd_table.chunk_size as bkgd_chunk_size " \
            "FROM raw_table JOIN bkgd_table ON raw_table.raw_id = bkgd_table.raw_id"

    table_result = t_env.sql_query(query)


    ds = t_env.to_append_stream(table_result,

        Types.ROW([<http://Types.INT|Types.INT>(), Types.STRING(), <http://Types.INT|Types.INT>(), <http://Types.INT|Types.INT>(), Types.STRING(), <http://Types.INT|Types.INT>(), <http://Types.INT|Types.INT>()]))

    with ds.execute_and_collect() as results:
        for result in results:
            print(result)

@Amir Hossein Sharifzadeh Yes, this is correct. After converting it to DataStream, you can use

ds.process

/

ds.flat_map

to perform the transformations you want to do in

process_data

.

@Dian Fu thank you, my process_data function has the same arguments as my ds:

(raw_id: int , raw_enc_val: string , raw_file_size: int, raw_chunk_size: int, bkgd_enc_val: string, bkgd_file_size: int, bkgd_chunk_size: int)

now, how should I pass

process_data

function to

ds.process

with those arguments?

It takes the whole row as input and you could refer to https://github.com/apache/flink/blob/master/flink-python/pyflink/examples/table/mixing_use_of_datastream_and_table.py#L63 as an example.

@Dian Fu So, here rather than

lambda i: i[0] + i[1]

I shall pass my function:

process_data

?

@Amir Hossein Sharifzadeh I’m not sure if have taken a look at the documentation of ds.process [1], it should be clearly documented that it accepts a ProcessFunction as inputs. [1] https://nightlies.apache.org/flink/flink-docs-release-1.16/api/python/reference/pyflink[…]api/pyflink.datastream.data_stream.DataStream.process.html

@Dian Fu I have already looked at process_function and it sounds like that we should pass a class rather than a function. There are not unfortunately enough examples to understand better

.process()

function.

As ProcessFunction provides many abilities e.g. accessing the current timestamp and so it needs to take a class as input. You could also use flat_map which allows to take function as input.

I am not sure if ProcessFunction address my question since I still don’t know how I would be able to pass tableresult into the ProcessFunction class, and how to call

process_results

into the class. Does

ds.flat_map(process_results)

sound like a solution?

You can call

ds.flat_map(process_results)

, see https://github.com/apache/flink/blob/master/flink-python/pyflink/examples/datastream/word_count.py#L89 for a simple examples about flat_map.