Running python job on Flink cluster hangs

Question

I have a problem submitting my python job to the Flink cluster (k8s).

I was able to successfully submit and run example job using command: ./bin/flink run --python examples/python/table/batch/word_count.py

However when I create my custom job the console freezes at:

./bin/flink run --python bin/flink.py
ERROR StatusLogger Reconfiguration failed: No configuration found for '5ce65a89' at 'null' in 'null'
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.flink.api.java.ClosureCleaner (file:/opt/flink/lib/apache-flink-examples-0.0.2-RC3.jar) to field java.util.Properties.serialVersionUID
WARNING: Please consider reporting this to the maintainers of org.apache.flink.api.java.ClosureCleaner
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release

nothing apperas on cluster (Job Manager) (word_count job was listed in Completed Job List).

My flink.py looks like this:

import logging
import sys

from pyflink.table import (EnvironmentSettings, TableEnvironment, DataTypes, StreamTableEnvironment)

from pyflink.datastream import StreamExecutionEnvironment

def main():
    env = StreamExecutionEnvironment.get_execution_environment()
    env.add_jars("file:///opt/lib/flink-connector-kafka_2.11-1.13.6.jar;file:///flink-json-1.13.6.jar,file:///flink-sql-connector-kafka_2.11-1.13.6.jar")
    t_env = StreamTableEnvironment.create(env)


    # source
    t_env.execute_sql("""
            CREATE TABLE trades (
              instrumentId INT,
              exchange STRING,
              category1 STRING,
              category2 STRING,
              traderId INT,
              price DOUBLE,
              quantity DOUBLE
            ) WITH (
              'connector' = 'kafka',
              'topic' = 'jse-trades',
              'properties.bootstrap.servers' = '',
              'properties.group.id' = 'flink-reader',
              'scan.startup.mode' = 'earliest-offset',
              'format' = 'json'
            )
        """)

    # sink
    t_env.execute_sql("""
            CREATE TABLE trades_aggr (
              instrumentId INT,
              exchange STRING,
              category1 STRING,
              category2 STRING,
              traderId INT,
              trade_count BIGINT,
              trade_turnover DOUBLE,
              trade_volume DOUBLE,
              trade_volume_avg DOUBLE,
              PRIMARY KEY (instrumentId) NOT ENFORCED
            ) WITH (
              'connector' = 'upsert-kafka',
              'topic' = 'jse-trades-results',
              'properties.bootstrap.servers' = '',
              'properties.group.id' = 'testGroup',
              'sink.buffer-flush.max-rows' = '5000',
              'sink.buffer-flush.interval' = '1000',
              'sink.parallelism' = '2',
              'key.format' = 'json',
              'value.format' = 'json'
            )
        """)

    # aggregation
    t_env.execute_sql("""
            INSERT INTO trades_aggr
                SELECT
                    instrumentId,
                    exchange,
                    category1,
                    category2,
                    traderId,
                    count(*),
                    sum(price*quantity),
                    sum(quantity),
                    avg(quantity)
                FROM trades
                GROUP BY instrumentId, exchange, category1, category2, traderId
        """)

    t_env.execute("kafka_trades")

if __name__ == "__main__":
    logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")

    main()

Equivalent Java job runs fine. It looks like I'm missing something..

Running python job on Flink cluster hangs

0 Answers