Just writing a simple code in pyflink that consumes data from a kafka topic and then transforms one of the columns in orders_sum sql_query. However I am not able to run this and gives error "Could not find any factory for identifier 'kafka' that implements 'org.apache.flink.table.factories.DynamicTableFactory' in the classpath."

def log_processing():

env_settings = <http://EnvironmentSettings.in|EnvironmentSettings.in>_streaming_mode()

t_env = TableEnvironment.create(env_settings)

t_env.get_config().set("pipeline.jars", "file:///flink-sql-connector-kafka-1.17.1.jar")

t_env.get_config().set("table.exec.source.idle-timeout", "1000")

source_ddl = """

CREATE TABLE restuarant_live_pending_orders(

rest_id VARCHAR,

status VARCHAR

) WITH (

'connector' = 'kafka',

'topic' = 'live_order_status',

'properties.bootstrap.servers' = 'localhost:9092',

'properties.group.id' = 'rest_group',

'scan.startup.mode' = 'specific-offsets',

'scan.startup.specific-offsets' = 'partition:0,offset:0',

'json.fail-on-missing-field' = 'false',

'json.ignore-parse-errors' = 'true',

'format' = 'json'

)

"""

t_env.execute_sql(source_ddl)

tbl = t_env.from_path('restuarant_live_pending_orders')

tbl.print_schema()

orders_sum = t_env.sql_query ("SELECT rest_id, SUM(CASE WHEN status = 'NEW' THEN 1 ELSE -1 END) AS status_count FROM %s GROUP BY rest_id" % tbl).execute()

orders_sum.print_schema()

@dp api A little strange for me. Could you check if the file

file:///flink-sql-connector-kafka-1.17.1.jar

really exists?

@Dian Fu - thank you for the response. yes, it was very odd. I changed the jar file location for it to work. Follow up query -> my rest_id is unique set of 100 restaurant IDs and i would like to print a dynamic table that is grouped by 'rest_id' (column1) and then COUNT of 'Status' (column2). Sample kafka ingest data:

{"rest_id": "6487f4c6fc13ae161d9008f6", "status": "NEW"}

So, the dynamic table is 100 x 2 table with changing COUNT of 'status' on 100 rest_ids. I am attempting to print this out on console first using Table API/SQL queries. Since, it a basic query I am sure I am using it correctly (like this

("SELECT rest_id, SUM(CASE WHEN status = 'NEW' THEN 1 ELSE -1 END) AS status_count FROM %s GROUP BY rest_id" % tbl)

) However, the output on console is printed with Updates/Inserts (see below) i.e. COUNT of Status is not being grouped by rest_id (as it keeps getting repeated)

8> -U[6487f4c6fc13ae161d9008f7, 3]

8> +U[6487f4c6fc13ae161d9008f7, 4]

7> -U[6487f4c6fc13ae161d9008e3, 3]

8> -U[6487f4c6fc13ae161d9008f7, 4]

7> +U[6487f4c6fc13ae161d9008e3, 4]

I believe I am missing something fundamental here. Do we have to mention a special sink (like upsert-kafka or some other connector) to output the dynamic table of 100 x 2 instead of print it w/o sink? if yes, which one to use?

However, the output on console is printed with Updates/Inserts (see below)

This is as expected.

-U

means retracting previous output

+U

means an updated row If you don’t want to see this, you can write the results to an output storage which supports upsert, e.g. MySQL, etc.

COUNT of Status is not being grouped by rest_id (as it keeps getting repeated)

Actually the results are grouped, you can see that the count is increasing. There is no repeating. It’s retraction message. I guess you could refer to the following documentation to understand some fundamental concepts: https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/concepts/versioned_tables/ https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/concepts/dynamic_tables/ https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/python/table/intro_to_table_api/#write-sql-queries

Hey Dian Thanks for the reply.. Since I am using python and want to see the UPSERT data in a table format, I am trying to connect to mysql server (running locally on docker) as a sink table. I have downloaded the JAR file for mysql version 8.0.32 from this link. However I am unable to set the configuration. Can a table environment have 2 JAR files config - one for kafka and one for mysql ? If so, can you share the syntax ?

def log_processing():

    env_settings = EnvironmentSettings.in_streaming_mode()
    t_env = TableEnvironment.create(env_settings)
    t_env.get_config().set("pipeline.jars", "file:///Users/Raghav/Desktop/prototype-v0.0.1/flink-sql-connector-kafka-1.17.1.jar", "file:///Users/Raghav/Desktop/prototype-v0.0.1/mysql-connector-j-8.0.32.jar")

    t_env.get_config().set("table.exec.source.idle-timeout", "1000")
    
    source_ddl = """
            CREATE TABLE restuarant_live_pending_orders(
                rest_id VARCHAR,
                status VARCHAR
            ) WITH (
              'connector' = 'kafka',
              'topic' = 'live_order_status',
              'properties.bootstrap.servers' = 'localhost:9092',
              'properties.group.id' = 'rest_group',
              'scan.startup.mode' = 'specific-offsets',
              'scan.startup.specific-offsets' = 'partition:0,offset:0',
              'json.fail-on-missing-field' = 'false',
              'json.ignore-parse-errors' = 'true',
              'format' = 'json'
            )
            """
    t_env.execute_sql(source_ddl)

    query = """
        INSERT INTO pending_orders_table
        SELECT rest_id, SUM(CASE WHEN status = 'NEW' THEN 1 WHEN status = 'PROCESSED' THEN -1 ELSE 0 END) AS pending_count
        FROM restuarant_live_pending_orders
        GROUP BY rest_id
    """
    
    sink_mysql = """
        CREATE TABLE pending_orders_table (
        rest_id VARCHAR,
        pending_count INT,
        PRIMARY KEY (rest_id) NOT ENFORCED
    ) WITH (
        'connector' = 'jdbc',
        'url' = 'jdbc:<mysql://0.0.0.0:3306/flink>',
        'table-name' = 'pending_orders_table',
        'username' = <username>,
        'password' = <password>
    )
    """
    t_env.execute_sql(sink_mysql)
    t_env.execute_sql(query)
    
if __name__ == '__main__':
    log_processing()

Thanks a ton for all your help!!

Yes, it supports multiple jars, see https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/python/dependency_management/#jar-dependencies for more details.

I have downloaded 2 JAR files from MAVEN central repo - 1. mysql-connector-j-8.0.32 -> driver dependency 2. flink-connector-jdbc-3.1.0-1.17 -> jdbc connector Do I need to configure both the JAR files in my script to connect to MySQL or should the first JAR file be enough ?

Both

from pyflink.table import EnvironmentSettings, TableEnvironment
from pyflink.table.expressions import *
from pyflink.table.table import Table

def log_processing():

    env_settings = EnvironmentSettings.in_streaming_mode()
    t_env = TableEnvironment.create(env_settings)
    t_env.get_config().set("pipeline.jars", "file:///Users/Raghav/Desktop/prototype-v0.0.1/flink-sql-connector-kafka-1.17.1.jar;file:///Users/Raghav/Desktop/prototype-v0.0.1/mysql-connector-j-8.0.32.jar;file:///Users/Raghav/Desktop/prototype-v0.0.1/flink-connector-jdbc-3.1.0-1.17.jar")
    t_env.get_config().set("table.exec.source.idle-timeout", "1000")
    
    source_ddl = """
            CREATE TABLE restuarant_live_pending_orders(
                rest_id VARCHAR,
                status VARCHAR
            ) WITH (
              'connector' = 'kafka',
              'topic' = 'live_order_status',
              'properties.bootstrap.servers' = 'localhost:9092',
              'properties.group.id' = 'rest_group',
              'scan.startup.mode' = 'specific-offsets',
              'scan.startup.specific-offsets' = 'partition:0,offset:0',
              'json.fail-on-missing-field' = 'false',
              'json.ignore-parse-errors' = 'true',
              'format' = 'json'
            )
            """
    t_env.execute_sql(source_ddl)

    sink_mysql = """
        CREATE TABLE pending_orders_table (
        rest_id VARCHAR,
        pending_count INT,
        PRIMARY KEY (rest_id) NOT ENFORCED
    ) WITH (
        'connector' = 'jdbc',
        'url' = 'jdbc:<mysql://localhost:3306/flink>',
        'table-name' = 'pending_orders_table',
        'username' = <USERNAME>,
        'password' = <PASSWORD>,
        'driver' = 'com.mysql.jdbc.Driver'
    )
    """
    t_env.execute_sql(sink_mysql)
    

    query = """
            INSERT INTO pending_orders_table
            SELECT rest_id, SUM(CASE WHEN status = 'NEW' THEN 1 WHEN status = 'PROCESSED' THEN -1 ELSE 0 END) AS pending_count
            FROM restuarant_live_pending_orders
            GROUP BY rest_id
        """
    t_env.execute_sql(query).wait()
    
if __name__ == '__main__':
    log_processing()

This is the code and now its giving no errors. All configs seem to be fine and I am able to connect to the mysql localhost server from a separate sql editor. However the flink sink query is not creating any table in mysql server under the database schema 'flink'. Since I am using .wait(), the code runs without giving any logs or any errors. Can you provide any suggestions on how to make this run ?

However the flink sink query is not creating any table in mysql server under the database schema ‘flink’.

This is by design. You need to create the table at mysql side yourself.

Have created a mysql table called flink.pending_orders_table where flink is the database name and pending_orders_table is table name. However I can still see no data in mysql. I cannot understand where my pipeline breaking.

Maybe you could configure

pipeline.operator-chaining: false

to disable the operator chain. In this way, you could see how many records are each operator received / sent from Flink Web UI.

I actually am not using Flink Web UI.. I only started using Flink a week ago and taking it one at a time. I am comfortable in Python hence prefer coding in here. Still not able to run my code. Is it mandatory to use JDBC catalog to send data into a table in MySQL ?

Great to hear. 🎉