我有一个非常简化的用例：我想使用Apache Flink(1.11(从一个Kafka主题中读取数据(我们称之为source_topic(，计算其中的一个属性(称为b(，并将结果写入另一个Kavka主题(result_topi(。
到目前为止，我有以下代码：
from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
from pyflink.table import StreamTableEnvironment, EnvironmentSettings
def log_processing():
env = StreamExecutionEnvironment.get_execution_environment()
env_settings = EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(stream_execution_environment=env, environment_settings=env_settings)`
t_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True)
t_env.get_config().get_configuration().set_string("pipeline.jars", "file:///opt/flink-1.11.2/lib/flink-sql-connector-kafka_2.11-1.11.2.jar")
source_ddl = """
CREATE TABLE source_table(
a STRING,
b INT
) WITH (
'connector' = 'kafka',
'topic' = 'source_topic',
'properties.bootstrap.servers' = 'node-1:9092',
'scan.startup.mode' = 'earliest-offset',
'format' = 'csv',
'csv.ignore-parse-errors' = 'true'
)
"""
sink_ddl = """
CREATE TABLE result_table(
b INT,
result BIGINT
) WITH (
'connector' = 'kafka',
'topic' = 'result_topic',
'properties.bootstrap.servers' = 'node-1:9092',
'format' = 'csv'
)
"""
t_env.execute_sql(source_ddl)
t_env.execute_sql(sink_ddl)
t_env.execute_sql("INSERT INTO result_table SELECT b,COUNT(b) FROM source_table GROUP BY b")
t_env.execute("Kafka_Flink_job")
if __name__ == '__main__':
log_processing()
但当我执行它时，我会得到以下错误：
py4j.protocol.Py4JJavaError: An error occurred while calling o5.executeSql.
: org.apache.flink.table.api.TableException: Table sink 'default_catalog.default_database.result_table' doesn't support consuming update changes which is produced by node GroupAggregate(groupBy=[b], select=[b, COUNT(b) AS EXPR$1])
我能够用一个简单的SELECT语句将数据写入Kafka主题。但是，一旦我添加了GROUP BY子句，就会抛出上面的异常。我遵循了Flink关于将Table API与SQL for Python一起使用的文档：https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/common.html#sql
非常感谢任何帮助，我是流处理和Flink的新手。非常感谢。
使用GROUP BY子句将生成更新流，而自Flink 1.11起，Kafka连接器不支持该更新流。另一方面，当您使用一个没有任何聚合的简单SELECT语句时，结果流只是追加的(这就是为什么您能够毫无问题地使用它(。
Flink 1.12即将发布，它包含了一个新的upstart Kafka连接器(如果你好奇的话，FLIP-149(，它将允许你也在PyFlink(即Python表API(中执行这种类型的操作。