26、Flink 的SQL之概览与入门示例

Flink 系列文章

一、Flink 专栏

Flink 专栏系统介绍某一知识点，并辅以具体的示例进行说明。

1、Flink 部署系列本部分介绍Flink的部署、配置相关基础内容。
2、Flink基础系列本部分介绍Flink 的基础部分，比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。
3、Flik Table API和SQL基础系列本部分介绍Flink Table Api和SQL的基本用法，比如Table API和SQL创建库、表用法、查询、窗口函数、catalog等等内容。
4、Flik Table API和SQL提高与应用系列本部分是table api 和sql的应用部分，和实际的生产应用联系更为密切，以及有一定开发难度的内容。
5、Flink 监控系列本部分和实际的运维、监控工作相关。

二、Flink 示例专栏

Flink 示例专栏是 Flink 专栏的辅助说明，一般不会介绍知识点的信息，更多的是提供一个一个可以具体使用的示例。本专栏不再分目录，通过链接即可看出介绍的内容。

两专栏的所有文章入口点击：Flink 系列文章汇总索引

文章目录

本文简单的介绍了SQL和SQL的入门，并以三个简单的示例进行介绍，由于示例涉及到其他的内容，需要了解更深入的内容请参考相关的文章。。
本文依赖flink和kafka、hadoop集群能正常使用。
本文分为2个部分，即介绍了Flink SQL和入门，并提供了完整的可验证通过的示例。

一、SQL

本文描述了 Flink 所支持的 SQL 语言，包括数据定义语言（Data Definition Language，DDL）、数据操纵语言（Data Manipulation Language，DML）以及查询语言。Flink 对 SQL 的支持基于实现了 SQL 标准的 Apache Calcite。

本文列出了目前（截至版本1.17） Flink SQL 所支持的所有语句：

SELECT (Queries)，具体内容参考文章：27、Flink 的SQL之SELECT (Queries)
CREATE TABLE, CATALOG, DATABASE, VIEW, FUNCTION 具体内容参考文章：22、Flink 的table api与sql之创建表的DDL
DROP TABLE, DATABASE, VIEW, FUNCTION
ALTER TABLE, DATABASE, FUNCTION
INSERT
ANALYZE TABLE 具体内容参考文章：28、Flink 的SQL之DROP 语句、ALTER 语句、INSERT 语句、ANALYZE 语句
UPDATE
DELETE
SQL HINTS
DESCRIBE
EXPLAIN
USE
SHOW
LOAD
UNLOAD

具体内容参考文章： 29、Flink SQL之DESCRIBE、EXPLAIN、USE、SHOW、LOAD、UNLOAD、SET、RESET、JAR、JOB Statements、UPDATE、DELETE

1、数据类型

通用类型与（嵌套的）复合类型（如：POJO、tuples、rows、Scala case 类) 都可以作为行的字段。

复合类型的字段任意的嵌套可被值访问函数（内置函数）访问。

通用类型将会被视为一个黑箱，且可以被用户自定义函数传递或引用。

对于 DDL 语句而言，我们支持所有在数据类型页面中定义的数据类型。

SQL查询不支持部分数据类型（cast 表达式或字符常量值）。
如：STRING, BYTES, RAW, TIME§ WITHOUT TIME ZONE, TIME§ WITH LOCAL TIME ZONE, TIMESTAMP§ WITHOUT TIME ZONE, TIMESTAMP§ WITH LOCAL TIME ZONE, ARRAY, MULTISET, ROW.

更多内容，请参考文章：14、Flink 的table api与sql之数据类型: 内置数据类型以及它们的属性

2、保留关键字

虽然 SQL 的特性并未完全实现，但是一些字符串的组合却已经被预留为关键字以备未来使用。如果你希望使用以下字符串作为你的字段名，请在使用时使用反引号将该字段名包起来（如

value

count

）。

A, ABS, ABSOLUTE,ACTION, ADA,ADD, ADMIN,AFTER,ALL, ALLOCATE, ALLOW,ALTER, ALWAYS,AND,ANALYZE,ANY, ARE, ARRAY,AS,ASC, ASENSITIVE, ASSERTION, ASSIGNMENT, ASYMMETRIC, AT, ATOMIC, ATTRIBUTE, ATTRIBUTES,AUTHORIZATION, AVG, BEFORE,BEGIN, BERNOULLI,BETWEEN,BIGINT,BINARY,BIT,BLOB,BOOLEAN, BOTH, BREADTH,BY, BYTES, C,CALL, CALLED, CARDINALITY,CASCADE,CASCADED,CASE, CAST, CATALOG, CATALOG_NAME, CEIL, CEILING, CENTURY,CHAIN,CHAR,CHARACTER, CHARACTERISTICS, CHARACTERS, CHARACTER_LENGTH, CHARACTER_SET_CATALOG, CHARACTER_SET_NAME, CHARACTER_SET_SCHEMA, CHAR_LENGTH,CHECK, CLASS_ORIGIN, CLOB,CLOSE,COALESCE, COBOL,COLLATE, COLLATION, COLLATION_CATALOG, COLLATION_NAME, COLLATION_SCHEMA, COLLECT,COLUMN,COLUMNS, COLUMN_NAME, COMMAND_FUNCTION, COMMAND_FUNCTION_CODE,COMMIT,COMMITTED, CONDITION, CONDITION_NUMBER,CONNECT, CONNECTION, CONNECTION_NAME,CONSTRAINT, CONSTRAINTS, CONSTRAINT_CATALOG, CONSTRAINT_NAME, CONSTRAINT_SCHEMA, CONSTRUCTOR,CONTAINS,CONTINUE,CONVERT, CORR, CORRESPONDING, COUNT, COVAR_POP, COVAR_SAMP,CREATE,CROSS, CUBE, CUME_DIST,CURRENT, CURRENT_CATALOG,CURRENT_DATE, CURRENT_DEFAULT_TRANSFORM_GROUP, CURRENT_PATH, CURRENT_ROLE, CURRENT_SCHEMA,CURRENT_TIME,CURRENT_TIMESTAMP, CURRENT_TRANSFORM_GROUP_FOR_TYPE,CURRENT_USER,CURSOR, CURSOR_NAME,CYCLE,DATA,DATABASE,DATE, DATETIME_INTERVAL_CODE, DATETIME_INTERVAL_PRECISION,DAY,DEALLOCATE,DEC, DECADE,DECIMAL,DECLARE,DEFAULT, DEFAULTS, DEFERRABLE, DEFERRED, DEFINED,DEFINER, DEGREE,DELETE, DENSE_RANK, DEPTH, DEREF, DERIVED,DESC,DESCRIBE, DESCRIPTION, DESCRIPTOR,DETERMINISTIC, DIAGNOSTICS, DISALLOW, DISCONNECT, DISPATCH,DISTINCT, DOMAIN,DOUBLE, DOW, DOY,DROP, DYNAMIC, DYNAMIC_FUNCTION, DYNAMIC_FUNCTION_CODE, EACH, ELEMENT,ELSE,END,END-EXEC, EPOCH, EQUALS,ESCAPE, EVERY,EXCEPT, EXCEPTION, EXCLUDE, EXCLUDING,EXEC,EXECUTE,EXISTS, EXP,EXPLAIN, EXTEND, EXTERNAL, EXTRACT,FALSE,FETCH, FILTER, FINAL,FIRST, FIRST_VALUE,FLOAT, FLOOR,FOLLOWING,FOR,FOREIGN, FORTRAN, FOUND, FRAC_SECOND, FREE,FROM,FULL,FUNCTION, FUSION, G, GENERAL, GENERATED, GET,GLOBAL, GO,GOTO,GRANT, GRANTED,GROUP, GROUPING,HAVING, HIERARCHY, HOLD,HOUR,IDENTITY, IMMEDIATE, IMPLEMENTATION,IMPORT,IN, INCLUDING, INCREMENT, INDICATOR, INITIALLY,INNER,INOUT, INPUT, INSENSITIVE,INSERT, INSTANCE, INSTANTIABLE,INT,INTEGER,INTERSECT, INTERSECTION,INTERVAL,INTO,INVOKER,IS,ISOLATION, JAVA,JOIN, K,KEY, KEY_MEMBER, KEY_TYPE, LABEL,LANGUAGE, LARGE,LAST, LAST_VALUE, LATERAL, LEADING,LEFT, LENGTH,LEVEL, LIBRARY,LIKE,LIMIT, LN,LOCAL, LOCALTIME, LOCALTIMESTAMP, LOCATOR, LOWER, M, MAP,MATCH,MATCHED, MAX, MAXVALUE, MEMBER,MERGE, MESSAGE_LENGTH, MESSAGE_OCTET_LENGTH, MESSAGE_TEXT, METHOD, MICROSECOND, MILLENNIUM, MIN,MINUTE, MINVALUE, MOD,MODIFIES, MODULE, MODULES,MONTH, MORE, MULTISET, MUMPS, NAME, NAMES,NATIONAL,NATURAL,NCHAR, NCLOB, NESTING, NEW,NEXT,NO, NONE, NORMALIZE, NORMALIZED,NOT,NULL, NULLABLE,NULLIF, NULLS, NUMBER,NUMERIC, OBJECT, OCTETS, OCTET_LENGTH,OF,OFFSET, OLD,ON, ONLY,OPEN,OPTION, OPTIONS,OR,ORDER, ORDERING, ORDINALITY, OTHERS,OUT,OUTER, OUTPUT,OVER, OVERLAPS, OVERLAY, OVERRIDING, PAD, PARAMETER, PARAMETER_MODE, PARAMETER_NAME, PARAMETER_ORDINAL_POSITION, PARAMETER_SPECIFIC_CATALOG, PARAMETER_SPECIFIC_NAME, PARAMETER_SPECIFIC_SCHEMA,PARTIAL,PARTITION, PASCAL, PASSTHROUGH, PATH, PERCENTILE_CONT, PERCENTILE_DISC, PERCENT_RANK, PLACING,PLAN, PLI, POSITION, POWER,PRECEDING,PRECISION,PREPARE, PRESERVE,PRIMARY, PRIOR,PRIVILEGES,PROCEDURE,PUBLIC, QUARTER, RANGE, RANK, RAW,READ,READS,REAL, RECURSIVE, REF,REFERENCES, REFERENCING, REGR_AVGX, REGR_AVGY, REGR_COUNT, REGR_INTERCEPT, REGR_R2, REGR_SLOPE, REGR_SXX, REGR_SXY, REGR_SYY, RELATIVE,RELEASE,REPEATABLE, RESET, RESTART,RESTRICT, RESULT,RETURN, RETURNED_CARDINALITY, RETURNED_LENGTH, RETURNED_OCTET_LENGTH, RETURNED_SQLSTATE,RETURNS,REVOKE,RIGHT, ROLE,ROLLBACK, ROLLUP,ROUTINE, ROUTINE_CATALOG, ROUTINE_NAME, ROUTINE_SCHEMA,ROW,ROWS, ROW_COUNT, ROW_NUMBER,SAVEPOINT, SCALE,SCHEMA, SCHEMA_NAME, SCOPE, SCOPE_CATALOGS, SCOPE_NAME, SCOPE_SCHEMA, SCROLL, SEARCH,SECOND, SECTION, SECURITY,SELECT, SELF, SENSITIVE, SEQUENCE,SERIALIZABLE, SERVER, SERVER_NAME,SESSION,SESSION_USER,SET, SETS, SIMILAR,SIMPLE, SIZE,SMALLINT,SOME, SOURCE, SPACE, SPECIFIC, SPECIFICTYPE, SPECIFIC_NAME,SQL, SQLEXCEPTION, SQLSTATE, SQLWARNING, SQL_TSI_DAY, SQL_TSI_FRAC_SECOND, SQL_TSI_HOUR, SQL_TSI_MICROSECOND, SQL_TSI_MINUTE, SQL_TSI_MONTH, SQL_TSI_QUARTER, SQL_TSI_SECOND, SQL_TSI_WEEK, SQL_TSI_YEAR, SQRT,START, STATE, STATEMENT, STATIC, STDDEV_POP, STDDEV_SAMP, STREAM, STRING, STRUCTURE, STYLE, SUBCLASS_ORIGIN, SUBMULTISET, SUBSTITUTE, SUBSTRING, SUM, SYMMETRIC, SYSTEM,SYSTEM_USER,TABLE, TABLESAMPLE, TABLE_NAME,TEMPORARY,THEN, TIES,TIME,TIMESTAMP, TIMESTAMPADD, TIMESTAMPDIFF, TIMEZONE_HOUR, TIMEZONE_MINUTE,TINYINT,TO, TOP_LEVEL_COUNT, TRAILING,TRANSACTION, TRANSACTIONS_ACTIVE, TRANSACTIONS_COMMITTED, TRANSACTIONS_ROLLED_BACK, TRANSFORM, TRANSFORMS, TRANSLATE, TRANSLATION, TREAT,TRIGGER, TRIGGER_CATALOG, TRIGGER_NAME, TRIGGER_SCHEMA, TRIM,TRUE,TYPE, UESCAPE,UNBOUNDED,UNCOMMITTED, UNDER,UNION,UNIQUE, UNKNOWN, UNNAMED, UNNEST,UPDATE, UPPER, UPSERT,USAGE,USER, USER_DEFINED_TYPE_CATALOG, USER_DEFINED_TYPE_CODE, USER_DEFINED_TYPE_NAME, USER_DEFINED_TYPE_SCHEMA,USING,VALUE,VALUES,VARBINARY,VARCHAR,VARYING, VAR_POP, VAR_SAMP, VERSION,VIEW, WEEK,WHEN, WHENEVER,WHERE, WIDTH_BUCKET, WINDOW,WITH,WITHIN, WITHOUT,WORK, WRAPPER,WRITE, XML,YEAR, ZONE

二、SQL入门

Flink SQL 使得使用标准 SQL 开发流应用程序变的简单。如果你曾经在工作中使用过兼容 ANSI-SQL 2011 的数据库或类似的 SQL 系统，那么就很容易学习 Flink。

1、Flink SQL环境准备

1）、安装Flink及提交任务方式

参考文章：
1、Flink1.12.7或1.13.5详细介绍及本地安装部署、验证
2、Flink1.13.5二种部署方式(Standalone、Standalone HA )、四种提交任务方式（前两种及session和per-job）验证详细步骤

2）、SQL客户端使用介绍

20、Flink SQL之SQL Client: 不用编写代码就可以尝试 Flink SQL，可以直接提交 SQL 任务到集群上

3）、简单示例

Flink SQL>SET execution.result-mode=tableau;

Flink SQL>showdatabases;+------------------+|database name |+------------------+| default_database |+------------------+1rowinset

Flink SQL>use default_database;[INFO]Execute statement succeed.

Flink SQL>showtables;
Empty set

Flink SQL>SELECT'Hello World';+----+--------------------------------+| op |                         _o__c0 |+----+--------------------------------+|+I |                    Hello World |+----+--------------------------------+
Received a total of1row
Flink SQL>show functions;
Hive Session ID =5d34cbf8-5984-4ec0-8527-e06a948ad7ca
+--------------------------------+|function name |+--------------------------------+|!||!=||                          $sum0 ||%||&||*||+||-||/||<||<=||<=>||<>||=||==||>||>=||                         IFNULL ||               SOURCE_WATERMARK ||                         TYPEOF ||^||           _legacy_grouping__id ||                            abs ||                           acos ||                     add_months ||                    aes_decrypt ||                    aes_encrypt ||and||                          array ||                 array_contains ||as||asc||                          ascii ||                           asin ||                    assert_true ||                assert_true_oom ||                             at ||                           atan ||                          atan2 ||                            avg ||                         base64 ||between||bigint||                            bin ||binary||                   bloom_filter ||boolean||                         bround ||                    cardinality ||          cardinality_violation ||case||                           cast ||                           cbrt ||                           ceil ||                        ceiling ||char||                     charLength ||                    char_length ||               character_length ||                            chr ||coalesce||                        collect ||                   collect_list ||                    collect_set ||                  compute_stats ||                         concat ||                      concat_ws ||                 context_ngrams ||                           conv ||                           corr ||                            cos ||                           cosh ||                            cot ||                          count ||                      covar_pop ||                     covar_samp ||                          crc32 ||                   create_union ||                    currentDate ||                   currentRange ||                     currentRow ||            currentRowTimestamp ||                    currentTime ||               currentTimestamp ||             current_authorizer ||               current_database ||                 current_groups ||current_user||date||                     dateFormat ||                       date_add ||                    date_format ||                       date_sub ||                       datediff ||day||                     dayofmonth ||                      dayofweek ||decimal||                         decode ||                        degrees ||desc||distinct||div||                         divide ||double||                              e ||                        element ||                            elt ||                         encode ||             encryptphonenumber ||end||             enforce_constraint ||                         equals ||                            exp ||                        explode ||                        extract ||                  extract_union ||                      factorial ||                          field ||                    find_in_set ||                        flatten ||float||                          floor ||                      floor_day ||                     floor_hour ||                   floor_minute ||                    floor_month ||                  floor_quarter ||                   floor_second ||                     floor_week ||                     floor_year ||                  format_number ||                     fromBase64 ||                  from_unixtime ||             from_utc_timestamp ||                            get ||                get_json_object ||                     get_splits ||                    greaterThan ||             greaterThanOrEqual ||                       greatest ||                       grouping ||hash||                            hex ||              histogram_numeric ||hour||if||                     ifThenElse ||in||                in_bloom_filter ||                        in_file ||index||                        initCap ||                        initcap ||                         inline ||                          instr ||int||              internal_interval ||              interval_day_time ||            interval_year_month ||                        isFalse ||                     isNotFalse ||                      isNotNull ||                      isNotTrue ||                         isNull ||                         isTrue ||                        isfalse ||                     isnotfalse ||                      isnotnull ||                      isnottrue ||                         isnull ||                         istrue ||                    java_method ||                     json_tuple ||                       last_day ||                          lcase ||                          least ||                         length ||                       lessThan ||                lessThanOrEqual ||                    levenshtein ||like||                        likeall ||                        likeany ||                             ln ||                      localTime ||                 localTimestamp ||                         locate ||                            log ||                          log10 ||                           log2 ||                 logged_in_user ||                          lower ||                      lowerCase ||                           lpad ||                          ltrim ||                            map ||                       map_keys ||                     map_values ||                           mask ||                   mask_first_n ||                      mask_hash ||                    mask_last_n ||              mask_show_first_n ||               mask_show_last_n ||                      matchpath ||                            max ||                            md5 ||                            min ||                          minus ||                    minusPrefix ||minute||                            mod ||month||                 months_between ||                    murmur_hash ||                   named_struct ||                       negative ||                       next_day ||                         ngrams ||                           noop ||                  noopstreaming ||                    noopwithmap ||           noopwithmapstreaming ||not||                     notBetween ||                      notEquals ||nullif||                            nvl ||                   octet_length ||or||over||                        overlay ||                      parse_url ||                parse_url_tuple ||                     percentile ||              percentile_approx ||                             pi ||                           plus ||                           pmod ||                     posexplode ||                       position ||                       positive ||                            pow ||                          power ||                         printf ||                       proctime ||                        quarter ||                        radians ||                           rand ||                    randInteger ||                        rangeTo ||                        reflect ||                       reflect2 ||regexp||                  regexpExtract ||                  regexpReplace ||                 regexp_extract ||                 regexp_replace ||                      regr_avgx ||                      regr_avgy ||                     regr_count ||                 regr_intercept ||                        regr_r2 ||                     regr_slope ||                       regr_sxx ||                       regr_sxy ||                       regr_syy ||                reinterpretCast ||repeat||replace||                 replicate_rows ||    restrict_information_schema ||                        reverse ||rlike||                          round ||row||                        rowtime ||                           rpad ||                          rtrim ||second||                      sentences ||                            sha ||                           sha1 ||                           sha2 ||                         sha224 ||                         sha256 |.................

至此，我们的环境都准备好了。

2、Source 表介绍及示例

与所有 SQL 引擎一样，Flink 查询操作是在表上进行。与传统数据库不同，Flink 不在本地管理静态数据；相反，它的查询在外部表上连续运行。

Flink 数据处理流水线开始于 source 表。source 表产生在查询执行期间可以被操作的行；它们是查询时 FROM 子句中引用的表。这些表可能是 Kafka 的 topics，数据库，文件系统，或者任何其它 Flink 知道如何消费的系统。

可以通过 SQL 客户端或使用环境配置文件来定义表。SQL 客户端支持类似于传统 SQL 的 SQL DDL 命令。标准 SQL DDL 用于创建，修改，删除表。

Flink 支持不同的连接器和格式相结合以定义表。相关内容在本Flink专栏中均有介绍，请参考：alanchanchn的专栏-Flink专栏

下面是一个示例，定义一个以 CSV 文件作为存储格式的 source 表。由于Flink创建表涉及较多的内容，关于下面的示例请参考文章：16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及FileSystem示例（1）

Flink SQL>show catalogs;
Hive Session ID =008f6263-1b8e-4eb7-b034-a2c8651809f1
+------------------+|     catalog name |+------------------+| alan_hivecatalog ||  default_catalog |+------------------+2rowsinset

Flink SQL>use catalog default_catalog;
Hive Session ID =1b1a3fb2-e303-4c2a-bfc8-5f38c47aa0f6
[INFO]Execute statement succeed.

Flink SQL>showdatabases;+------------------+|database name |+------------------+| default_database |+------------------+1rowinset

Flink SQL>use default_database;[INFO]Execute statement succeed.

Flink SQL>showtables;
Empty set

Flink SQL>CREATETABLE alan_first_table (>     t_id BIGINT,>     t_name STRING,>     t_balance DOUBLE,>     t_age INT>)WITH(>'connector'='filesystem',>'path'='hdfs://HadoopHAcluster/flinktest/firstdemo/',>'format'='csv'>);[INFO]Execute statement succeed.

Flink SQL>showtables;+------------------+|table name |+------------------+| alan_first_table |+------------------+1rowinset---能查出来数据是有前提的，那就是在创建表之前，我已经在hdfs://HadoopHAcluster/flinktest/firstdemo目录下上传了5个文件，每个文件一条数据[alanchan@server4 testdata]$ hadoop fs -ls hdfs://HadoopHAcluster/flinktest/firstdemo
Found 5 items
-rw-r--r--   3 alanchan supergroup         15 2023-09-07 10:24 hdfs://HadoopHAcluster/flinktest/firstdemo/dim_user1.txt-rw-r--r--   3 alanchan supergroup         19 2023-09-07 10:24 hdfs://HadoopHAcluster/flinktest/firstdemo/dim_user2.txt-rw-r--r--   3 alanchan supergroup         22 2023-09-07 10:24 hdfs://HadoopHAcluster/flinktest/firstdemo/dim_user3.txt-rw-r--r--   3 alanchan supergroup         20 2023-09-07 10:24 hdfs://HadoopHAcluster/flinktest/firstdemo/dim_user4.txt-rw-r--r--   3 alanchan supergroup         24 2023-09-07 10:24 hdfs://HadoopHAcluster/flinktest/firstdemo/dim_user5.txt

Flink SQL>select*from alan_first_table;+----+----------------------+--------------------------------+--------------------------------+-------------+| op |                 t_id |                         t_name |                      t_balance |       t_age |+----+----------------------+--------------------------------+--------------------------------+-------------+|+I |5|                  alan_chan_chn |52.23|38||+I |3|                    alanchanchn |32.23|28||+I |1|                           alan |12.23|18||+I |4|                      alan_chan |12.43|29||+I |2|                       alanchan |22.23|10|+----+----------------------+--------------------------------+--------------------------------+-------------+
Received a total of5rows---带条件查询
Flink SQL>select*from alan_first_table where t_balance >=20;+----+----------------------+--------------------------------+--------------------------------+-------------+| op |                 t_id |                         t_name |                      t_balance |       t_age |+----+----------------------+--------------------------------+--------------------------------+-------------+|+I |3|                    alanchanchn |32.23|28||+I |2|                       alanchan |22.23|10||+I |5|                  alan_chan_chn |52.23|38|+----+----------------------+--------------------------------+--------------------------------+-------------+
Received a total of3rows

可以从该表中定义一个连续查询，当新行可用时读取并立即输出它们的结果。

3、连续查询介绍及示例

虽然最初设计时没有考虑流语义，但 SQL 是用于构建连续数据流水线的强大工具。Flink SQL 与传统数据库查询的不同之处在于，Flink SQL 持续消费到达的行并对其结果进行更新。

一个连续查询永远不会终止，并会产生一个动态表作为结果。动态表是 Flink 中 Table API 和 SQL 对流数据支持的核心概念。

连续流上的聚合需要在查询执行期间不断地存储聚合的结果。例如，假设你需要从传入的数据流中计算每个部门的员工人数。查询需要维护每个部门最新的计算总数，以便在处理新行时及时输出结果。

关于连续查询更多的内容，参考文章：15、Flink 的table api与sql之流式概念-详解的介绍了动态表、时间属性配置（如何处理更新结果）、时态表、流上的join、流上的确定性以及查询配置

下面的示例说明：
1、在flink创建一张表，提交连续查询的任务（其实就是一个查询session，动态显示表内的数据）
2、为方便模拟，使用kafka作为消息源，即表的连接类型为kafka，也即需要有kafka的运行环境
3、sql客户端的环境与本文上述示例一致
4、关于该示例更多的信息参考：16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Kafka示例（3）

Flink SQL>CREATETABLE alanchan_kafka_table (>`id`INT,>     name STRING,>     age INT,>     balance DOUBLE>)WITH(>'connector'='kafka',>'topic'='t_kafka_source',>'scan.startup.mode'='earliest-offset',>'properties.bootstrap.servers'='192.168.10.41:9092,192.168.10.42:9092,192.168.10.43:9092',>'format'='csv'>);[INFO]Execute statement succeed.

Flink SQL>showtables;+----------------------+|table name |+----------------------+|     alan_first_table || alanchan_kafka_table |+----------------------+2rowsinset-----kafka一条一条写入数据，下文中的查询结果会根据kafka中发送的消息逐条展示出来------[alanchan@server2 bin]$ kafka-console-producer.sh --broker-list server1:9092 --topic t_kafka_source>1,alan,15,100>2,alanchan,20,200>3,alanchanchn,25,300>4,alan_chan,30,400>5,alan_chan_chn,50,45>

Flink SQL>select*from alanchan_kafka_table;+----+-------------+--------------------------------+-------------+--------------------------------+| op |          id |                           name |         age |                        balance |+----+-------------+--------------------------------+-------------+--------------------------------+|+I |1|                           alan |15|100.0||+I |2|                       alanchan |20|200.0||+I |3|                    alanchanchn |25|300.0||+I |4|                      alan_chan |30|400.0||+I |5|                  alan_chan_chn |50|45.0|

4、Sink 表介绍及示例

当运行此查询时，SQL 客户端实时但是以只读方式提供输出。存储结果，作为报表或仪表板的数据来源，需要写到另一个表。这可以使用 INSERT INTO 语句来实现。本节中引用的表称为 sink 表。INSERT INTO 语句将作为一个独立查询被提交到 Flink 集群中。

------创建数据源表，该表不能查询
Flink SQL>CREATETABLE source_table (>  userId INT,>  age INT,>  balance DOUBLE,>  userName STRING
>)WITH(>'connector'='datagen',>'rows-per-second'='100',>'fields.userId.kind'='sequence',>'fields.userId.start'='1',>'fields.userId.end'='1000',>>'fields.balance.kind'='random',>'fields.balance.min'='1',>'fields.balance.max'='100',>>'fields.age.min'='1',>'fields.age.max'='1000',>>'fields.userName.length'='10'>);[INFO]Execute statement succeed.----创建sink表，hdfs文件夹不需要手动创建，flink会自己创建
Flink SQL>CREATETABLE alan_sink_table (>     t_id BIGINT,>     t_name STRING,>     t_balance DOUBLE,>     t_age INT>)WITH(>'connector'='filesystem',>'path'='hdfs://HadoopHAcluster/flinktest/firstsinkdemo/',>'format'='csv'>);[INFO]Execute statement succeed.------批量插入sink表，也可以是动态的，但需要设置数据刷新频率，否则查不到结果，该事情在本Flink专栏中有说明------此处也是提交一个flink任务，此处用的是yarn-session模式
Flink SQL>INSERTINTO alan_sink_table 
>SELECT userId ,userName,balance,age FROM source_table;

Job ID: c2e1985745c5c938c56e26f8efe5a8db

------查询结果如下
Flink SQL>select*from alan_sink_table;+----+----------------------+--------------------------------+--------------------------------+-------------+| op |                 t_id |                         t_name |                      t_balance |       t_age |+----+----------------------+--------------------------------+--------------------------------+-------------+|+I |1|                     d0c7d38b94 |31.52935530019297|802||+I |2|                     b880adc262 |45.43292342494475|556||+I |3|                     e1ce373b2e |39.595138772111014|459||+I |4|3bd1242679 |78.58761035208113|585||+I |5|88ba47bb2b |4.870598793833649|508||+I |6|72bdba9132 |48.33565877511729|115||+I |7|0fa82976d1 |52.6978279057911|353||+I |8|8d546bab93 |20.403401648898576|391||+I |9|9eb957d512 |82.16967630094122|323||+I |10|5423755f01 |49.12646233699912|769||+I |11|                     da6c7936ea |16.877530563314846|687||+I |12|3ef87eb75a |68.65154273578702|434||+I |13|                     e08320e927 |8.403066874855323|292||+I |14|03e1ccfc69 |98.61326426348097|653|......+----+----------------------+--------------------------------+--------------------------------+-------------+
Received a total of1000rows

提交后，它将运行并将结果直接存储到 sink 表中，而不是将结果加载到系统内存中。

以上，简单的介绍了SQL和SQL的入门，并以三个简单的示例进行介绍，由于示例涉及到其他的内容，需要了解更深入的内容请参考相关的文章。

标签： flink sql 大数据

本文转载自: https://blog.csdn.net/chenwewi520feng/article/details/132039813
版权归原作者 一瓢一瓢的饮 alanchanchn 所有，如有侵权，请联系我们删除。