大数据流处理批处理数据工程【免费下载链接】flink项目地址https://gitcode.com/gh_mirrors/fli/flink点击查看免费下载本指南基于 Apache Flink 的 Hive 方言Hive Dialect能力系统讲解在使用 Hive 语法编写 SQL 时支持的 5 类 CREATE 语句CREATE DATABASE、CREATE TABLE、CREATE VIEW、CREATE MACRO与CREATE FUNCTION。你将掌握每类语句的完整语法、关键参数语义、与 Hive 原生语法的兼容边界并结合仓库源码与集成测试理解其底层实现与最佳实践从而在 Flink 与 Hive 之间无缝迁移建表/建库脚本。前置准备如何切换到 Hive 方言Hive 方言从 Flink 1.11.0 开始引入目的是让用户直接使用 Hive 语法编写 SQL改善与 Hive 的互操作性减少在 Flink 与 Hive 之间来回切换执行不同语句的成本。Flink 目前支持default与hive两种 SQL 方言且可以在会话中按语句动态切换无需重启会话详见 Hive 方言概览。在进入 CREATE 语句讲解之前先确认以下前提必须已添加 Hive 相关依赖参考 Hive connector 的 dependencies 章节确保当前 Catalog 是 HiveCatalog否则将回落到 Flink 默认方言强烈建议加载 HiveModule 并将其置于 Module 列表首位以便函数解析时优先使用 Hive 内置函数Hive 方言只支持db.table两级标识符不支持带 Catalog 名的标识符部分特性是否可用取决于实际使用的 Hive 版本例如更新数据库位置仅 Hive 2.4.0 支持Hive 方言主要面向批模式部分语法如 Sort/Cluster/Distribute By、Transform在流模式下尚未支持。三种设置方言的方式# SQL Client通过 table.sql-dialect 属性动态切换 Flink SQL SET table.sql-dialect hive; -- 使用 Hive 方言 [INFO] Session property has been set. Flink SQL SET table.sql-dialect default; -- 使用 Flink 默认方言 [INFO] Session property has been set.# 启动 HiveServer2 Endpoint 的 SQL Gateway 默认即 Hive 方言可直接使用 jdbc:hive2 SET table.sql-dialect default; -- 切回 Flink 默认方言 jdbc:hive2 SET table.sql-dialect hive; -- 切回 Hive 方言// Table API (Java)通过 TableConfig 设置方言 EnvironmentSettings settings EnvironmentSettings.inStreamingMode(); TableEnvironment tableEnv TableEnvironment.create(settings); tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE); // 使用 Hive 方言 tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT); // 使用默认方言# Table API (Python) from pyflink.table import * settings EnvironmentSettings.in_batch_mode() t_env TableEnvironment.create(settings) t_env.get_config().set_sql_dialect(SqlDialect.HIVE) t_env.get_config().set_sql_dialect(SqlDialect.DEFAULT)CREATE DATABASE功能说明CREATE DATABASE或等价的CREATE SCHEMA用于创建指定名称的数据库。语法CREATE (DATABASE|SCHEMA) [IF NOT EXISTS] database_name [COMMENT database_comment] [LOCATION hdfs_path] [WITH DBPROPERTIES (property_nameproperty_value, ...)];各子句语义IF NOT EXISTS目标数据库已存在时静默跳过不报错COMMENT为数据库添加注释LOCATION指定数据库在文件系统如 HDFS上的存储路径若不指定则落到 Hive 默认 warehouse 下WITH DBPROPERTIES为数据库附加键值对元数据可通过DESCRIBE DATABASE EXTENDED查看。示例CREATE DATABASE db1; CREATE DATABASE IF NOT EXISTS db1 COMMENT db1 LOCATION /user/hive/warehouse/db1 WITH DBPROPERTIES (nameexample-db);从源码实现看CREATE DATABASE属于 DDL 节点集合DDL_NODES中的TOK_CREATEDATABASE见 HiveParser.java其执行结果通过 HiveCatalog 写入 Hive Metastore。集成测试 HiveDialectITCase.testCreateDatabase 验证了create database db1 comment db1 comment后可通过hiveCatalog.getHiveDatabase(db1)读到注释LOCATION与DBPROPERTIES均被正确持久化locationUri与parameters.get(k1)与语句完全一致。CREATE TABLE功能说明CREATE TABLE用于在已有数据库中定义一张表包括管理表与外部表。注意Hive 方言目前不支持创建临时表。语法CREATE [EXTERNAL] TABLE [IF NOT EXISTS] [db_name.]table_name [(col_name data_type [column_constraint] [COMMENT col_comment], ... [table_constraint])] [COMMENT table_comment] [PARTITIONED BY (col_name data_type [COMMENT col_comment], ...)] [ [ROW FORMAT row_format] [STORED AS file_format] ] [LOCATION fs_path] [TBLPROPERTIES (property_nameproperty_value, ...)] [AS select_statment];支持的数据类型data_type : primitive_type | array_type | map_type | struct_type primitive_type : TINYINT | SMALLINT | INT | BIGINT | BOOLEAN | FLOAT | DOUBLE | DOUBLE PRECISION | STRING | BINARY | TIMESTAMP | DECIMAL | DECIMAL(precision, scale) | DATE | VARCHAR | CHAR array_type : ARRAY data_type struct_type : STRUCT col_name : data_type [COMMENT col_comment], ...其中map_type与array_type结构一致MAP key_type, value_type 、ARRAY data_type 支持与 Hive 相同的嵌套组合。行格式ROW FORMATrow_format: : DELIMITED [FIELDS TERMINATED BY char [ESCAPED BY char]] [COLLECTION ITEMS TERMINATED BY char] [MAP KEYS TERMINATED BY char] [LINES TERMINATED BY char] [NULL DEFINED AS char] | SERDE serde_name [WITH SERDEPROPERTIES (property_nameproperty_value, ...)]DELIMITED形式用于文本类分隔符配置字段分隔符FIELDS TERMINATED BY、转义字符ESCAPED BY、集合元素分隔符COLLECTION ITEMS TERMINATED BY、Map 键值分隔符MAP KEYS TERMINATED BY、行分隔符LINES TERMINATED BY以及空值表示NULL DEFINED ASSERDE形式直接指定 SerDe 类名并可通过WITH SERDEPROPERTIES传入 SerDe 参数。文件格式STORED ASfile_format: : SEQUENCEFILE | TEXTFILE | RCFILE | ORC | PARQUET | AVRO | INPUTFORMAT input_format_classname OUTPUTFORMAT output_format_classname内置支持SEQUENCEFILE、TEXTFILE、RCFILE、ORC、PARQUET、AVRO六种常见格式也可通过INPUTFORMAT ... OUTPUTFORMAT ...指定自定义的 InputFormat/OutputFormat 类。列约束与表约束column_constraint: : NOT NULL table_constraint: : [CONSTRAINT constraint_name] PRIMARY KEY (col_name, ...)示例-- 创建非分区表 CREATE TABLE t1(key string, value string); -- 创建分区表 CREATE TABLE pt1(key string, value string) PARTITIONED BY (year int, month int); -- 指定存储格式创建表 CREATE TABLE t1(key string, value string) STORED AS ORC; -- 指定行格式创建表自定义分隔符 CREATE TABLE t1(m MAPBIGINT, STRING) ROW FORMAT DELIMITED COLLECTION ITEMS TERMINATED BY ; MAP KEYS TERMINATED BY :; -- CTAS根据查询结果建表 CREATE TABLE t2 AS SELECT key, COUNT(1) FROM t1 GROUP BY key;源码与测试验证集成测试 HiveDialectITCase.testCreateTable 对上述各分支做了完整验证外部表CREATE EXTERNAL TABLE tbl1 ... LOCATION ... TBLPROPERTIES(k1v1)创建后Metastore 中表类型为EXTERNAL_TABLE分区键数量、Location、TBLPROPERTIES 均与语句一致存储格式STORED AS ORC会同时设置OrcSerde、OrcInputFormat与OrcOutputFormat自定义 SerDeROW FORMAT SERDE ...LazyBinarySerDe会写入对应serializationLib文本分隔符FIELDS TERMINATED BY |会同时写入 SerDe 参数的field.delim与serialization.formatLINES TERMINATED BY写入line.delimCOLLECTION ITEMS TERMINATED BY与MAP KEYS TERMINATED BY分别写入collection.delim与mapkey.delimIF NOT EXISTS重复执行建表不会覆盖已有表表创建时间不变约束在 Hive 3.1.0 上NOT NULL ... RELY会反映到 Flink Schema 的 nullable 属性NORELY则不会PRIMARY KEY (x) DISABLE RELY会被解析为带约束名的主键见 testCreateTableWithConstraintsCTASCREATE TABLE ... AS SELECT会依据查询结果的列名与类型建表且STORED AS指定的格式会被继承见 testCreateTableAs。CREATE VIEW功能说明CREATE VIEW创建指定名称的视图。若未显式提供列名视图列名将由 SELECT 表达式自动推导当 SELECT 包含未命名的标量表达式如x y时生成的列名为_C0、_C1等。重命名列时也可同时提供列注释注释不会自动从底层列继承。视图是纯逻辑对象不关联任何存储。查询引用视图时会先求值视图定义产生行集再交给外层查询继续处理。语法CREATE VIEW [IF NOT EXISTS] [db_name.]view_name [(column_name, ...) ] [COMMENT view_comment] [TBLPROPERTIES (property_name property_value, ...)] AS SELECT ...;示例CREATE VIEW IF NOT EXISTS v1 (key COMMENT key) COMMENT View for key1 AS SELECT key FROM src WHERE key 1;该示例创建视图v1显式声明列key并加注释视图本身带注释TBLPROPERTIES 可省略定义来自src表按key 1过滤后的结果。CREATE MACRO功能说明CREATE TEMPORARY MACRO使用给定的可选列列表作为表达式输入来创建一个宏。宏只存在于当前会话期间session 级别会话结束即失效。语法CREATE TEMPORARY MACRO macro_name([col_name col_type, ...]) expression;示例-- 无参宏 CREATE TEMPORARY MACRO fixed_number() 42; -- 单参数宏字符串长度加 2 CREATE TEMPORARY MACRO string_len_plus_two(x string) length(x) 2; -- 多参数宏 CREATE TEMPORARY MACRO simple_add (x int, y int) x y;在 HiveParser.java 中TOK_CREATEMACRO与TOK_DROPMACRO均被纳入 DDL 节点集合说明宏的创建/删除走的是 Hive 语法解析路径。同时测试代码还约束了宏命名的合法性宏名中不允许出现.字符见 HiveDialectITCase否则会抛出 CREATE TEMPORARY MACRO doesnt allow . character in the macro name 的错误。CREATE FUNCTION功能说明CREATE FUNCTION创建由指定类class_name实现的函数支持临时函数与永久函数两种形态并可通过USING JAR子句携带函数实现及其依赖的 Jar 包。语法创建临时函数函数仅存在于当前会话期间CREATE TEMPORARY FUNCTION function_name AS class_name [USING JAR file_uri];创建永久函数函数注册到 Metastore除非显式 DROP否则在所有会话中均存在CREATE FUNCTION [db_name.]function_name AS class_name [USING JAR file_uri];参数说明USING JAR file_uri该子句用于在创建函数时附带包含函数实现及其依赖的 Jar 包file_uri可以是本地文件路径也可以是分布式文件系统如 HDFS上的路径当函数在查询中被实际使用时Flink 会自动为远程 Jar 执行下载下载的 Jar 会在会话退出时被清理。示例-- 类 SimpleUdf 已存在于 classpath直接创建函数 CREATE FUNCTION simple_udf AS SimpleUdf; -- 类未在 classpath通过本地 Jar 提供实现 CREATE FUNCTION simple_udf AS SimpleUdf USING JAR /tmp/SimpleUdf.jar; -- 类未在 classpath通过 HDFS 远程 Jar 提供实现 CREATE FUNCTION simple_udf AS SimpleUdf USING JAR hdfs://namenode-host:port/path/SimpleUdf.jar;集成测试 testCreateFunctionUsingJar 验证了该流程的端到端可用性测试动态编译一个继承org.apache.hadoop.hive.ql.exec.UDF、实现evaluate(int)加一的 UDF 类打成 Jar然后分别用CREATE FUNCTION ... USING JAR与CREATE TEMPORARY FUNCTION ... USING JAR注册再通过SELECT add_one(x) FROM src查询结果正确返回[I[2], I[3]]证明 Jar 中的类在会话内可被正确加载与调用。底层实现Hive 方言的解析链路Hive 方言的解析器通过工厂机制注册到 Flink 的 Parser SPI 中。在 HiveParserFactory.java 中factoryIdentifier()返回SqlDialect.HIVE.name().toLowerCase()即hive与table.sql-dialect hive对应create(Context)将上下文强转为CalciteContext并构造HiveParser因为 Hive 解析器需要借助 CalciteContext 构建 Calcite 的 RelNode 逻辑计划。HiveParser内部维护了一个庞大的DDL_NODES集合见 HiveParser.java其中就包含本文涉及的TOK_CREATEDATABASE、TOK_CREATETABLE、TOK_CREATEVIEW、TOK_CREATEFUNCTION、TOK_CREATEMACRO等 Hive AST 节点。这从源码层面印证了Hive 方言并非在 Flink 默认语法之上做字符串替换而是直接复用 Hive 自带的语法解析器生成 AST再将其翻译为 Flink 的 Catalog 操作与逻辑计划从而最大程度保证与 Hive 语法的兼容性。相关语句速查掌握 CREATE 之后配套的 Hive 方言语句还包括ALTER 语句修改数据库、表、视图的属性/分区/SerDe 等DROP 语句删除数据库、表、视图、函数、宏SHOW 语句展示数据库、表、分区、函数等信息INSERT 语句 与 LOAD DATA 语句向 Hive 表写入或加载数据SET 语句配置 Hive 相关会话变量。结合 HiveDialectITCase.java 中 1300 余行覆盖建库、建表、约束、CTAS、INSERT/OVERWRITE、函数、宏等场景的集成测试可以确信在当前仓库所对应的 Flink 版本中上述 CREATE 语句族在批模式下已具备与 Hive 高度一致的语法与行为适合作为将 Hive 存量 DDL 迁移到 Flink 的可靠依据。赞分享大数据流处理批处理数据工程【免费下载链接】flink项目地址https://gitcode.com/gh_mirrors/fli/flink点击查看免费下载相关推荐Flink Hive 方言 DROP 语句完全指南数据库、表、视图、宏与函数的删除操作Flink Hive 方言 DROP 语句完全指南数据库、表、视图、宏与函数的删除操作 导读 本文系统讲解 Apache Flink 在启用 Hive 方言后大数据流处理批处理数据工程Flink Hive Dialect ALTER 语句完全指南数据库、表与视图的元数据变更实战Flink Hive Dialect ALTER 语句完全指南数据库、表与视图的元数据变更实战 本文是 Flink Hive Dialect 系列语法文档之一大数据流处理批处理数据工程Flink SQL DROP 语句完全指南删除 Catalog、表、数据库、视图与函数的语法与实现原理Flink SQL DROP 语句完全指南删除 Catalog、表、数据库、视图与函数的语法与实现原理 在 Flink Table SQL 编程体系中D大数据流处理批处理数据工程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考