用Python在Flink中查询嵌套行
用Python在Flink中查询嵌套行
提问于 2022-05-30 21:13:31
基于pyflink walkthrough，我现在尝试使用apache-flink==1.14.4获得一个简单的嵌套行查询。我基于以下解决方案创建了我的表结构：Get nested fields from Kafka message using Apache Flink SQL
有一条消息是这样的：
{"signature": {"token": "abcd1234"}}代码的相关部分如下：
create_kafka_source_ddl = """
CREATE TABLE nested_msg (
`signature` ROW (
`token` STRING
)
) WITH (
'connector' = 'kafka',
'topic' = 'nested_msg',
'properties.bootstrap.servers' = 'kafka:9092',
'properties.group.id' = 'nested-msg',
'scan.startup.mode' = 'latest-offset',
'format' = 'json'
)
"""
create_es_sink_ddl = """
CREATE TABLE es_sink (
token STRING
) WITH (
'connector' = 'elasticsearch-7',
'hosts' = 'http://elasticsearch:9200',
'index' = 'nested_count_1',
'document-id.key-delimiter' = '$',
'sink.bulk-flush.max-size' = '42mb',
'sink.bulk-flush.max-actions' = '32',
'sink.bulk-flush.interval' = '1000',
'sink.bulk-flush.backoff.delay' = '1000',
'format' = 'json'
)
"""
t_env.execute_sql(create_kafka_source_ddl)
t_env.execute_sql(create_es_sink_ddl)
# How do I select the nested field here?
t_env.from_path("nested_msg").select(col("signature.token").alias("token")).select(
"token"
).execute_insert("es_sink")我在这里尝试过无数种变体，但都没有成功。例外是：
py4j.protocol.Py4JJavaError: An error occurred while calling o48.select.
: org.apache.flink.table.api.ValidationException: Cannot resolve field [signature.token], input field list:[signature].如何选择这样的嵌套字段以将其插入到我的接收器中？
回答 1
Stack Overflow用户
回答已采纳
发布于 2022-06-01 01:36:48
您可以将col("signature.token")更改为col("signature").get('token')。
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/72440086
复制相关文章
相似问题
腾讯云开发者
