Hello I m testing a simple aggregation function using a pyth Apache Flink #troubleshooting

Hello, I'm testing a simple aggregation function ...

Jaehyeon Kim

02/28/2023, 2:11 AM

Hello, I'm testing a simple aggregation function using a python udf. It obtains the number of words and characters in values by key. It also concatenates value strings. It works fine for print sink but fails for kafka sink with the following error.

Copy code

pyflink.util.exceptions.TableException: org.apache.flink.table.api.TableException: Table sink 'default_catalog.default_database.output' doesn't support consuming update changes which is produced by node PythonGroupAggregate(groupBy=[k], select=[k, *org.apache.flink.table.functions.python.PythonAggregateFunction$82f60ecbb5d8eed484a7f66dd2ee2921*(k, v) AS TMP_0])

Below show the print sink output.

Copy code

# input messages
a:a
b:hello
b:world
b:hi

# print
1> +I[a, 1, 1,  a]
1> +I[b, 2, 10,  hello world]
1> -U[b, 2, 10,  hello world]
1> +U[b, 3, 12,  hello world hi]

The script and full error messages are attached. Can you please guide me how to fix it?

s5_03_aggregation.py aggregation_error.txt

Dian Fu

02/28/2023, 3:04 AM

Hey @Jaehyeon Kim, You are using raw format which only accepts

append only

change logs. However, the aggregation node may produce

update changes

and so this is unsupported. The following format supports accepting

update changes

, perhaps you could give them a try: • debezium: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/table/formats/debezium/ • canal: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/table/formats/canal/ • maxwell: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/table/formats/maxwell/ • ogg: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/table/formats/ogg/

Dian Fu

02/28/2023, 3:19 AM

Note that you could also use

upsert-kafka

which accepts

update changes

. See https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/table/upsert-kafka/ for more details.

Jaehyeon Kim

02/28/2023, 3:26 AM

Thank you @Dian Fu. I'll try to implement with

upsert-kafka

Jaehyeon Kim

02/28/2023, 4:36 AM

It works with

upsert-kafka

. Here is the updated output table. Note that I had to set the key field name to be longer than 1 (

kk

). Also

raw

format doesn't accept multiple columns so that I changed the value format to

json

. Thanks @Dian Fu !

Copy code

table_env.execute_sql(
    f"""
    CREATE TABLE output (
        kk VARCHAR,
        wc BIGINT,
        cc BIGINT,
        w VARCHAR,
        PRIMARY KEY (kk) NOT ENFORCED
    ) WITH (
        'connector' = 'upsert-kafka',
        'topic' = 'aggregations-output-topic',
        'properties.bootstrap.servers' = '{BOOTSTRAP_SERVERS}',
        'key.format' = 'raw',
        'key.fields-prefix' = 'k',
        'value.format' = 'json',
        'value.json.fail-on-missing-field' = 'false',
        'value.fields-include' = 'EXCEPT_KEY'
    )
    """
)

👏 3

Raghunadh Nittala

03/08/2023, 3:19 PM

Hi @Jaehyeon Kim, I have a similar scenario where I’m creating table (with ‘upsert-kafka’ connector) and have a SQL query on top of this. When I use tableEnv.sqlQuery(), I’m getting the above error, With tableEnv.executeSql(), I’m able to print the results. Did you by any chance convert the tableResult to a Datastream or Table?

Jaehyeon Kim

03/08/2023, 10:36 PM

I haven't tried that. Probably it'd be better to create a new question to everyone.

3 Views

Open in Slack

Previous Next