You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by Zhonghan Tang <13...@163.com> on 2020/07/13 12:53:19 UTC
回复： flink 1.11运算结果存mysql出错
有新数据进来吗，看起来和这个jira很像
https://issues.apache.org/jira/browse/FLINK-15262
在2020年07月13日 20:38，Leonard Xu<xb...@gmail.com> 写道：
Hi,
简单看了下代码应该没啥问题，alarm_test_g 这个kafka topic里有数据吗？可以检查下是否有脏数据，直接用./bin/kafka-console-consumer.sh 检查下？我有点怀疑这点
Best,
Leonard Xu
在 2020年7月13日，20:06，小学生 <20...@qq.com> 写道：
各位大佬好，请教一个问题flink从Kafka读数，写入mysql，程序没有报错，但是没有写入mysql任何数据。代码如下，是在linux下，直接python *.py执行的。完整代码如下
from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic, CheckpointingMode
from pyflink.table import StreamTableEnvironment, EnvironmentSettings
source="""
CREATE TABLE kafka_source_tab (
id VARCHAR,
alarm_id VARCHAR,
trck_id VARCHAR
) WITH (
'connector' = 'kafka',
'topic' = 'alarm_test_g',
'scan.startup.mode' = 'earliest-offset',
'properties.bootstrap.servers' = '10.2.2.73:2181',
'properties.bootstrap.servers' = '10.2.2.73:9092',
'format' = 'json'
)
"""
sink="""
CREATE TABLE g_source_tab (
id VARCHAR,
alarm_id VARCHAR,
trck_id VARCHAR
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://10.2.2.77:3306/bdt?useSSL=false',
'table-name' = 'g',
'username' = 'root',
'password' = '123456t',
'sink.buffer-flush.interval' = '1s'
)
"""
env = StreamExecutionEnvironment.get_execution_environment()
env.set_stream_time_characteristic(TimeCharacteristic.EventTime)
env.set_parallelism(1)
env_settings = EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(env, environment_settings=env_settings)
t_env.execute_sql(source)
t_env.execute_sql(sink)
source = t_env.from_path("kafka_source_tab")\
.select("id,alarm_id,trck_id")
source.execute_insert("g_source_tab")
Re: flink 1.11运算结果存mysql出错
Posted by 小学生 <20...@qq.com>.
懂了，谢谢
Re: flink 1.11运算结果存mysql出错
Posted by godfrey he <go...@gmail.com>.
1.11 对 StreamTableEnvironment.execute()
和 StreamExecutionEnvironment.execute() 的执行方式有所调整，
简单概述为：
1. StreamTableEnvironment.execute() 只能执行 sqlUpdate 和 insertInto 方法执行作业；
2. Table 转化为 DataStream 后只能通过 StreamExecutionEnvironment.execute() 来执行作业；
3. 新引入的 TableEnvironment.executeSql() 和 StatementSet.execute() 方法是直接执行sql作业
(异步提交作业)，不需要再调用 StreamTableEnvironment.execute()
或 StreamExecutionEnvironment.execute()
TableEnvironment.executeSql() 和 StatementSet.execute()
提交的作业都是异步的，如果是在本地测试的话，不会等有最终结果才会推出。针对这个问题，1.12里准备引入 await 方法
[3]，代码还在review中。
TableResult是用来描述一个statement执行的结果。对于SELECT和INSERT，TableResult中还包含了JobClient
[4]
用来操作对应的job，例如获取job状态，cancel作业，等待作业结束等。TableResult还可以collect方法拿到statement执行的schema和结果数据，例如
select/show的结果。
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/common.html#%E7%BF%BB%E8%AF%91%E4%B8%8E%E6%89%A7%E8%A1%8C%E6%9F%A5%E8%AF%A2
[2]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/common.html#%E5%B0%86%E8%A1%A8%E8%BD%AC%E6%8D%A2%E6%88%90-datastream-%E6%88%96-dataset
[3] https://issues.apache.org/jira/browse/FLINK-18337
[4]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-74%3A+Flink+JobClient+API
Best,
Godfrey
小学生 <20...@qq.com> 于2020年7月13日周一 下午9:12写道：
> 嗯嗯，尝试了，这下没问题了，想问下这个TableResult对象，设计的目的是啥呢，不是特别懂呢，谢谢！
>
Re: flink 1.11运算结果存mysql出错
Posted by 小学生 <20...@qq.com>.
嗯嗯，尝试了，这下没问题了，想问下这个TableResult对象，设计的目的是啥呢，不是特别懂呢，谢谢！
Re: flink 1.11运算结果存mysql出错
Posted by Leonard Xu <xb...@gmail.com>.
Hi，知道了
source.execute_insert("g_source_tab”) 返回的结果是一个TableResult对象，如果不显示地等待任务的执行，这个任务会直接返回，你试下这个
result.execute_insert("g_source_tab") \
.get_job_client() \
.get_job_execution_result() \
.result()
这是Flip-84引入的一个改动，为了更好地处理table程序的返回值。
祝好，
Leonard Xu
> 在 2020年7月13日，20:57，小学生 <20...@qq.com> 写道：
>
> 不像吧，这个是1.10版的，我执行这个程序很快就结束了，不会挂着。
Re: flink 1.11运算结果存mysql出错
Posted by 小学生 <20...@qq.com>.
不像吧，这个是1.10版的，我执行这个程序很快就结束了，不会挂着。