在单个INSERT INTO ... SELECT ...
作业中(在Flink 1.13.6上)是否可能有多个PyFlink语句?
我创建了许多输出表,并试图在单个作业中写入这些表,其中Python & SQL示例(假设有一个名为“input”的输入表):
sql1 = "INSERT INTO out1 (col1, col2) SELECT col1, col2 FROM input"
sql2 = "INSERT INTO out2 (col3, col4) SELECT col3, col4 FROM input"
env.execute_sql(sql1)
env.execute_sql(sql2)
当这在AWS上的Kinesis中的Flink集群中运行时,我会得到一个失败:
在单个环境中不能有多个execute()或executeAsync()调用。
当我查看Flink web时,我可以看到有一个叫做insert-into_default_catalog.default_database.out1
的作业。Flink是否将每个INSERT语句分离到一个单独的作业中?它似乎试图为第一个查询创建一个作业,然后却无法为第二个查询创建第二个作业。
是否有任何方法可以让它使用SQL作为一个单一作业运行,而不必离开SQL和Table?
发布于 2022-04-29 14:47:19
如果要执行多个插入,则需要将它们包装在语句集中:
stmt_set = table_env.create_statement_set()
# only single INSERT query can be accepted by `add_insert_sql` method
stmt_set.add_insert_sql(sql1)
stmt_set.add_insert_sql(sql2)
# execute all statements together
table_result = stmt_set.execute()
# get job status through TableResult
print(table_result.get_job_client().get_job_status())
有关更多信息,请参见医生们。
https://stackoverflow.com/questions/72058541
复制相似问题