平顶山网站建设电话,app怎么开发制作,郑州网站建设老牌公司,朝阳网站关键词优化在《0基础学习PyFlink——使用PyFlink的SQL进行字数统计》一文中#xff0c;我们直接执行了Select查询操作#xff0c;在终端中直接看到了查询结果。
select word, count(1) as count from source group by word;
------------------------------------------------------
|…在《0基础学习PyFlink——使用PyFlink的SQL进行字数统计》一文中我们直接执行了Select查询操作在终端中直接看到了查询结果。
select word, count(1) as count from source group by word;
------------------------------------------------------
| word | count |
------------------------------------------------------
| A | 3 |
| B | 1 |
| C | 2 |
| D | 2 |
| E | 1 |
------------------------------------------------------在生产环境我们往往要将计算结果保存到外部系统中比如Mysql等。这个时候我们就要使用Sink。
Sink
Sink用于将Reduce结果输出到外部系统。它也是通过一个表Table来表示结构。这个和MapReduce思路中的Map很类似。
Print
为了简单起见我们让Sink的表连接的外部系统是print。这样我们就可以在控制台上看到数据。 # define the sinkmy_sink_ddl CREATE TABLE WordsCountTableSink (word STRING,count BIGINT,PRIMARY KEY (word) NOT ENFORCED) WITH (connector print);t_env.execute_sql(my_sink_ddl).print()这一步只能创建表和连接器具体执行还要执行下一步
Execute
因为source和WordsCountTableSink是两张表分别表示数据的输入和输出结构。如果要打通输入和输出则需要将source表中的数据通过某些计算插入到WordsCountTableSink表中。于是我们主要使用的是insert into指令。 # execute insertmy_select_ddl insert into WordsCountTableSinkselect word, count(1) as countfrom sourcegroup by wordt_env.execute_sql(my_select_ddl).wait()完整代码如下
import argparse
import logging
import sysfrom pyflink.common import Configuration
from pyflink.table import (EnvironmentSettings, TableEnvironment)def word_count(input_path):config Configuration()# write all the data to one fileconfig.set_string(parallelism.default, 1)env_settings EnvironmentSettings \.new_instance() \.in_batch_mode() \.with_configuration(config) \.build()t_env TableEnvironment.create(env_settings)# define the sourcemy_source_ddl create table source (word STRING) with (connector filesystem,format csv,path {}).format(input_path)t_env.execute_sql(my_source_ddl).print()tab t_env.from_path(source)# define the sinkmy_sink_ddl CREATE TABLE WordsCountTableSink (word STRING,count BIGINT,PRIMARY KEY (word) NOT ENFORCED) WITH (connector print);t_env.execute_sql(my_sink_ddl).print()# execute insertmy_select_ddl insert into WordsCountTableSinkselect word, count(1) as countfrom sourcegroup by wordt_env.execute_sql(my_select_ddl).wait()if __name__ __main__:logging.basicConfig(streamsys.stdout, levellogging.INFO, format%(message)s)parser argparse.ArgumentParser()parser.add_argument(--input,destinput,requiredFalse,helpInput file to process.)argv sys.argv[1:]known_args, _ parser.parse_known_args(argv)word_count(known_args.input)执行命令如下
python sql_print.py --input input1.csv输出结果如下 Using Any for unsupported type: typing.Sequence[~T] No module named google.cloud.bigquery_storage_v1. As a result, the ReadFromBigQuery transform CANNOT be used with methodDIRECT_READ. OK OK I[A, 3] I[B, 1] I[C, 2] I[D, 2] I[E, 1] 因为使用的是批处理模式in_batch_mode我们看到Flink将所有数据计算完整成成批的执行了新增操作代表新增。这块对比我们将在后续将流处理时介绍区别。 附上input1.csv内容
A,
B,
C,
D,
A,
E,
C,
D,
A,
