文章详情

登录

短信预约-IT技能 免费直播动态提醒

选择考试省份

  • 北京
  • 上海
  • 天津
  • 重庆
  • 河北
  • 山东
  • 辽宁
  • 黑龙江
  • 吉林
  • 甘肃
  • 青海
  • 河南
  • 江苏
  • 湖北
  • 湖南
  • 江西
  • 浙江
  • 广东
  • 云南
  • 福建
  • 海南
  • 山西
  • 四川
  • 陕西
  • 贵州
  • 安徽
  • 广西
  • 内蒙
  • 西藏
  • 新疆
  • 宁夏
  • 兵团

请输入下面的图形验证码

提交验证

短信预约提醒成功

一文了解基于Flink1.12构建流批一体数仓的技术点

2024-12-03 02:38

关注

本文转载自微信公众号「大数据技术与数仓」，作者西贝。转载本文请联系大数据技术与数仓公众号。

基于Flink构建流批一体的实时数仓是目前数据仓库领域比较火的实践方案。随着Flink的不断迭代，其提供的一系列技术特性使得用户构建流批一体的应用变得越来越方便。本文将以Flink1.12为例，一一介绍这些特性的基本使用方式，主要包括以下内容：

Flink集成Hive

使用Hive构建数据仓库已经成为了比较普遍的一种解决方案。目前，一些比较常见的大数据处理引擎，都无一例外兼容Hive。Flink从1.9开始支持集成Hive，不过1.9版本为beta版，不推荐在生产环境中使用。在Flink1.10版本中，标志着对 Blink的整合宣告完成，对 Hive 的集成也达到了生产级别的要求。值得注意的是，不同版本的Flink对于Hive的集成有所差异，本文将以最新的Flink1.12版本为例，阐述Flink集成Hive的简单步骤，以下是全文，希望对你有所帮助。

Flink集成Hive的基本方式

Flink 与 Hive 的集成主要体现在以下两个方面:

Flink利用 Hive 的 MetaStore 作为持久化的 Catalog，我们可通过HiveCatalog将不同会话中的 Flink 元数据存储到 Hive Metastore 中。例如，我们可以使用HiveCatalog将其 Kafka的数据源表存储在 Hive Metastore 中，这样该表的元数据信息会被持久化到Hive的MetaStore对应的元数据库中，在后续的 SQL 查询中，我们可以重复使用它们。

Flink打通了与Hive的集成，如同使用SparkSQL或者Impala操作Hive中的数据一样，我们可以使用Flink直接读写Hive中的表。

HiveCatalog的设计提供了与 Hive 良好的兼容性，用户可以”开箱即用”的访问其已有的 Hive表。不需要修改现有的 Hive Metastore，也不需要更改表的数据位置或分区。

Flink集成Hive的步骤

Flink支持的Hive版本
大版本 V1 V2 V3 V4 V5 V6 V7
1.0 1.0.0 1.0.1          
1.1 1.1.0 1.1.1          
1.2 1.2.0 1.2.1 1.2.2        
2.0 2.0.0 2.0.1          
2.1 2.1.0 2.1.1          
2.2 2.2.0            
2.3 2.3.0 2.3.1 2.3.2 2.3.3 2.3.4 2.3.5 2.3.6
3.1 3.1.0 3.1.1 3.1.2    

值得注意的是，对于不同的Hive版本，可能在功能方面有所差异，这些差异取决于你使用的Hive版本，而不取决于Flink，一些版本的功能差异如下：

依赖项

本文以Flink1.12为例，集成的Hive版本为Hive2.3.4。集成Hive需要额外添加一些依赖jar包，并将其放置在Flink安装目录下的lib文件夹下，这样我们才能通过 Table API 或 SQL Client 与 Hive 进行交互。

另外，Apache Hive 是基于 Hadoop 之上构建的, 所以还需要 Hadoop 的依赖，配置好HADOOP_CLASSPATH即可。这一点非常重要，否则在使用FlinkSQL Cli查询Hive中的表时，会报如下错误：

  1. java.lang.ClassNotFoundException: org.apache.hadoop.mapred.JobConf 

配置HADOOP_CLASSPATH，需要在/etc/profile文件中配置如下的环境变量：

  1. export HADOOP_CLASSPATH=`hadoop classpath` 

Flink官网提供了两种方式添加Hive的依赖项。第一种是使用 Flink 提供的 Hive Jar包(根据使用的 Metastore 的版本来选择对应的 Hive jar)，建议优先使用Flink提供的Hive jar包，这种方式比较简单方便。本文使用的就是此种方式。当然，如果你使用的Hive版本与Flink提供的Hive jar包兼容的版本不一致，你可以选择第二种方式，即别添加每个所需的 jar 包。

下面列举了可用的jar包及其适用的Hive版本，我们可以根据使用的Hive版本，下载对应的jar包即可。比如本文使用的Hive版本为Hive2.3.4，所以只需要下载flink-sql-connector-hive-2.3.6即可，并将其放置在Flink安装目录的lib文件夹下。

Metastore version Maven dependency SQL Client JAR
1.0.0 ~ 1.2.2 flink-sql-connector-hive-1.2.2 Download
2.0.0 ~2.2.0 flink-sql-connector-hive-2.2.0 Download
2.3.0 ~2.3.6 flink-sql-connector-hive-2.3.6 Download
3.0.0 ~ 3.1.2 flink-sql-connector-hive-3.1.2 Download

上面列举的jar包，是我们在使用Flink SQL Cli所需要的jar包，除此之外，根据不同的Hive版本，还需要添加如下jar包。以Hive2.3.4为例，除了上面的一个jar包之外，还需要添加下面两个jar包：

flink-connector-hive_2.11-1.12.0.jar和hive-exec-2.3.4.jar。其中hive-exec-2.3.4.jar包存在于Hive安装路径下的lib文件夹。flink-connector-hive_2.11-1.12.0.jar的下载地址为：

  1. https://repo1.maven.org/maven2/org/apache/flink/flink-connector-hive_2.11/1.12.0/ 

NOTE:black_nib::Flink1.12集成Hive只需要添加如下三个jar包，以Hive2.3.4为例，分别为：

flink-sql-connector-hive-2.3.6

flink-connector-hive_2.11-1.12.0.jar

hive-exec-2.3.4.jar

Flink SQL Cli集成Hive

将上面的三个jar包添加至Flink的lib目录下之后，就可以使用Flink操作Hive的数据表了。以FlinkSQL Cli为例：

配置sql-client-defaults.yaml

该文件时Flink SQL Cli启动时使用的配置文件，该文件位于Flink安装目录的conf/文件夹下，具体的配置如下，主要是配置catalog：

除了上面的一些配置参数，Flink还提供了下面的一些其他配置参数：

参数 必选 默认值 类型 描述
type 是 (无) String Catalog 的类型。创建 HiveCatalog 时，该参数必须设置为'hive'。
name 是 (无) String Catalog 的名字。仅在使用 YAML file 时需要指定。
hive-conf-dir 否 (无) String 指向包含 hive-site.xml 目录的 URI。该 URI 必须是 Hadoop 文件系统所支持的类型。如果指定一个相对 URI，即不包含 scheme，则默认为本地文件系统。如果该参数没有指定，我们会在 class path 下查找hive-site.xml。
default-database 否 default String 当一个catalog被设为当前catalog时，所使用的默认当前database。
hive-version 否 (无) String HiveCatalog 能够自动检测使用的 Hive 版本。我们建议不要手动设置 Hive 版本，除非自动检测机制失败。
hadoop-conf-dir 否 (无) String Hadoop 配置文件目录的路径。目前仅支持本地文件系统路径。我们推荐使用 HADOOP_CONF_DIR 环境变量来指定 Hadoop 配置。因此仅在环境变量不满足您的需求时再考虑使用该参数，例如当您希望为每个 HiveCatalog 单独设置 Hadoop 配置时。

操作Hive中的表

首先启动FlinkSQL Cli，命令如下：

  1. ./bin/sql-client.sh embedded 

接下来，我们可以查看注册的catalog

  1. Flink SQL> show catalogs; 
  2. default_catalog 
  3. myhive 

使用注册的myhive catalog

  1. Flink SQL> use catalog myhive; 

假设Hive中有一张users表，在Hive中查询该表：

  1. hive (default)> select * from users; 
  2. OK 
  3. users.id        users.mame 
  4. 1       jack 
  5. 2       tom 
  6. 3       robin 
  7. 4       haha 
  8. 5       haha 

查看对应的数据库表，我们可以看到Hive中已经存在的表，这样就可以使用FlinkSQL操作Hive中的表，比如查询，写入数据。

  1. Flink SQL> show tables; 
  2. Flink SQL> select * from users; 

向Hive表users中插入一条数据：

  1. Flink SQL> insert into users select 6,'bob'; 

再次使用Hive客户端去查询该表的数据，会发现写入了一条数据。

接下来，我们再在FlinkSQL Cli中创建一张kafka的数据源表：

  1. CREATE TABLE user_behavior (  
  2.     `user_id` BIGINT, -- 用户id 
  3.     `item_id` BIGINT, -- 商品id 
  4.     `cat_id` BIGINT, -- 品类id 
  5.     `action` STRING, -- 用户行为 
  6.     `province` INT, -- 用户所在的省份 
  7.     `ts` BIGINT, -- 用户行为发生的时间戳 
  8.     `proctime` AS PROCTIME(), -- 通过计算列产生一个处理时间列 
  9.     `eventTime` AS TO_TIMESTAMP(FROM_UNIXTIME(ts, 'yyyy-MM-dd HH:mm:ss')), -- 事件时间 
  10.      WATERMARK FOR eventTime AS eventTime - INTERVAL '5' SECOND  -- 定义watermark 
  11.  ) WITH (  
  12.     'connector' = 'kafka', -- 使用 kafka connector 
  13.     'topic' = 'user_behavior', -- kafka主题 
  14.     'scan.startup.mode' = 'earliest-offset', -- 偏移量 
  15.     'properties.group.id' = 'group1', -- 消费者组 
  16.     'properties.bootstrap.servers' = 'kms-2:9092,kms-3:9092,kms-4:9092',  
  17.     'format' = 'json', -- 数据源格式为json 
  18.     'json.fail-on-missing-field' = 'true', 
  19.     'json.ignore-parse-errors' = 'false' 
  20. ); 

查看表结构

  1. Flink SQL> DESCRIBE user_behavior; 

我们可以在Hive的客户端中执行下面命令查看刚刚在Flink SQLCli中创建的表

  1. hive (default)> desc formatted  user_behavior; 
  2. # Detailed Table Information              
  3. Database:               default                   
  4. Owner:                  null                      
  5. CreateTime:             Sun Dec 20 16:04:59 CST 2020      
  6. LastAccessTime:         UNKNOWN                   
  7. Retention:              0                         
  8. Location:               hdfs://kms-1.apache.com:8020/user/hive/warehouse/user_behavior    
  9. Table Type:             MANAGED_TABLE             
  10. Table Parameters:                 
  11.         flink.connector         kafka                
  12.         flink.format            json                 
  13.         flink.json.fail-on-missing-field        true                 
  14.         flink.json.ignore-parse-errors  false                
  15.         flink.properties.bootstrap.servers      kms-2:9092,kms-3:9092,kms-4:9092 
  16.         flink.properties.group.id       group1               
  17.         flink.scan.startup.mode earliest-offset      
  18.         flink.schema.0.data-type        BIGINT               
  19.         flink.schema.0.name     user_id              
  20.         flink.schema.1.data-type        BIGINT               
  21.         flink.schema.1.name     item_id              
  22.         flink.schema.2.data-type        BIGINT               
  23.         flink.schema.2.name     cat_id               
  24.         flink.schema.3.data-type        VARCHAR(2147483647)  
  25.         flink.schema.3.name     action               
  26.         flink.schema.4.data-type        INT                  
  27.         flink.schema.4.name     province             
  28.         flink.schema.5.data-type        BIGINT               
  29.         flink.schema.5.name     ts                   
  30.         flink.schema.6.data-type        TIMESTAMP(3) NOT NULL 
  31.         flink.schema.6.expr     PROCTIME()           
  32.         flink.schema.6.name     proctime             
  33.         flink.schema.7.data-type        TIMESTAMP(3)         
  34.         flink.schema.7.expr     TO_TIMESTAMP(FROM_UNIXTIME(`ts`, 'yyyy-MM-dd HH:mm:ss')) 
  35.         flink.schema.7.name     eventTime            
  36.         flink.schema.watermark.0.rowtime        eventTime            
  37.         flink.schema.watermark.0.strategy.data-type     TIMESTAMP(3)         
  38.         flink.schema.watermark.0.strategy.expr  `eventTime` - INTERVAL '5' SECOND 
  39.         flink.topic             user_behavior        
  40.         is_generic              true                 
  41.         transient_lastDdlTime   1608451499           
  42.                   
  43. # Storage Information             
  44. SerDe Library:          org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe        
  45. InputFormat:            org.apache.hadoop.mapred.TextInputFormat          
  46. OutputFormat:           org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat    
  47. Compressed:             No                        
  48. Num Buckets:            -1                        
  49. Bucket Columns:         []                        
  50. Sort Columns:           []                        
  51. Storage Desc Params:              
  52.         serialization.format    1              

NOTE:black_flag::在Flink中创建一张表，会把该表的元数据信息持久化到Hive的metastore中，我们可以在Hive的metastore中查看该表的元数据信息

进入Hive的元数据信息库，本文使用的是MySQL。执行下面的命令：

  1. SELECT  
  2.     a.tbl_id, -- 表id 
  3.     from_unixtime(create_time) AS create_time, -- 创建时间 
  4.     a.db_id, -- 数据库id 
  5.     b.name AS db_name, -- 数据库名称 
  6.     a.tbl_name -- 表名称 
  7. FROM TBLS AS a 
  8. LEFT JOIN DBS AS b ON a.db_id =b.db_id 
  9. WHERE a.tbl_name = "user_behavior"; 

使用代码连接到 Hive

maven依赖

  1. -- Flink Dependency --> 
  2.  
  3.   org.apache.flink 
  4.   flink-connector-hive_2.11 
  5.   1.12.0 
  6.  
  7.  
  8.   org.apache.flink 
  9.   flink-table-api-java-bridge_2.11 
  10.   1.12.0 
  11.  
  12. -- Hive Dependency --> 
  13.  
  14.     org.apache.hive 
  15.     hive-exec 
  16.     2.3.4 
  17.  

代码

  1. public class HiveIntegrationDemo { 
  2.     public static void main(String[] args) { 
  3.         EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().build(); 
  4.         TableEnvironment tableEnv = TableEnvironment.create(settings); 
  5.  
  6.         String name            = "myhive"; 
  7.         String defaultDatabase = "default"; 
  8.         String hiveConfDir = "/opt/modules/apache-hive-2.3.4-bin/conf"; 
  9.          
  10.         HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir); 
  11.         tableEnv.registerCatalog("myhive", hive); 
  12.         // 使用注册的catalog 
  13.         tableEnv.useCatalog("myhive"); 
  14.         // 向Hive表中写入一条数据  
  15.         String insertSQL = "insert into users select 10,'lihua'"; 
  16.  
  17.         TableResult result2 = tableEnv.executeSql(insertSQL); 
  18.         System.out.println(result2.getJobClient().get().getJobStatus()); 
  19.  
  20.     } 
  21. } 

提交程序，观察Hive表的变化：

  1. bin/flink run -m kms-1:8081 \ 
  2. -c com.flink.sql.hiveintegration.HiveIntegrationDemo \ 
  3. ./original-study-flink-sql-1.0-SNAPSHOT.jar 

Hive Catalog与Hive Dialect

什么是Hive Catalog

我们知道，Hive使用Hive Metastore(HMS)存储元数据信息，使用关系型数据库来持久化存储这些信息。所以，Flink集成Hive需要打通Hive的metastore，去管理Flink的元数据，这就是Hive Catalog的功能。

Hive Catalog的主要作用是使用Hive MetaStore去管理Flink的元数据。Hive Catalog可以将元数据进行持久化，这样后续的操作就可以反复使用这些表的元数据，而不用每次使用时都要重新注册。如果不去持久化catalog，那么在每个session中取处理数据，都要去重复地创建元数据对象，这样是非常耗时的。

如何使用Hive Catalog

HiveCatalog是开箱即用的，所以，一旦配置好Flink与Hive集成，就可以使用HiveCatalog。比如，我们通过FlinkSQL 的DDL语句创建一张kafka的数据源表，立刻就能查看该表的元数据信息。

HiveCatalog可以处理两种类型的表：一种是Hive兼容的表，另一种是普通表(generic table)。其中Hive兼容表是以兼容Hive的方式来存储的，所以，对于Hive兼容表而言，我们既可以使用Flink去操作该表，又可以使用Hive去操作该表。

普通表是对Flink而言的，当使用HiveCatalog创建一张普通表，仅仅是使用Hive MetaStore将其元数据进行了持久化，所以可以通过Hive查看这些表的元数据信息(通过DESCRIBE FORMATTED命令)，但是不能通过Hive去处理这些表，因为语法不兼容。

对于是否是普通表，Flink使用is_generic属性进行标识。默认情况下，创建的表是普通表，即is_generic=true，如果要创建Hive兼容表，需要在建表属性中指定is_generic=false。

尖叫提示：

由于依赖Hive Metastore，所以必须开启Hive MetaStore服务

代码中使用Hive Catalog

  1. EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().build(); 
  2.        TableEnvironment tableEnv = TableEnvironment.create(settings); 
  3.  
  4.        String name            = "myhive"; 
  5.        String defaultDatabase = "default"; 
  6.        String hiveConfDir = "/opt/modules/apache-hive-2.3.4-bin/conf"; 
  7.  
  8.        HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir); 
  9.        tableEnv.registerCatalog("myhive", hive); 
  10.        // 使用注册的catalog 
  11.        tableEnv.useCatalog("myhive"); 

Flink SQLCli中使用Hive Catalog

在FlinkSQL Cli中使用Hive Catalog很简单，只需要配置一下sql-cli-defaults.yaml文件即可。配置内容如下：

  1. catalogs: 
  2.    - name: myhive 
  3.      type: hive 
  4.      default-database: default 
  5.      hive-conf-dir: /opt/modules/apache-hive-2.3.4-bin/conf 

在FlinkSQL Cli中创建一张kafka表，该表默认为普通表，即is_generic=true

  1. CREATE TABLE user_behavior (  
  2.     `user_id` BIGINT, -- 用户id 
  3.     `item_id` BIGINT, -- 商品id 
  4.     `cat_id` BIGINT, -- 品类id 
  5.     `action` STRING, -- 用户行为 
  6.     `province` INT, -- 用户所在的省份 
  7.     `ts` BIGINT, -- 用户行为发生的时间戳 
  8.     `proctime` AS PROCTIME(), -- 通过计算列产生一个处理时间列 
  9.     `eventTime` AS TO_TIMESTAMP(FROM_UNIXTIME(ts, 'yyyy-MM-dd HH:mm:ss')), -- 事件时间 
  10.      WATERMARK FOR eventTime AS eventTime - INTERVAL '5' SECOND  -- 定义watermark 
  11.  ) WITH (  
  12.     'connector' = 'kafka', -- 使用 kafka connector 
  13.     'topic' = 'user_behavior', -- kafka主题 
  14.     'scan.startup.mode' = 'earliest-offset', -- 偏移量 
  15.     'properties.group.id' = 'group1', -- 消费者组 
  16.     'properties.bootstrap.servers' = 'kms-2:9092,kms-3:9092,kms-4:9092',  
  17.     'format' = 'json', -- 数据源格式为json 
  18.     'json.fail-on-missing-field' = 'true', 
  19.     'json.ignore-parse-errors' = 'false' 
  20. ); 

我们可以在Hive客户端中查看该表的元数据信息

  1. hive (default)> desc formatted  user_behavior; 
  2. Table Parameters:                 
  3.        ... 
  4.         is_generic              true                 
  5.       ...          

从上面的元数据信息可以看出，is_generic=true，说明该表是一张普通表，如果在Hive中去查看该表，则会报错。

上面创建的表是普通表，该表不能使用Hive去查询。那么，该如何创建一张Hive兼容表呢?我们只需要在建表的属性中显示指定is_generic=false即可，具体如下：

  1. CREATE TABLE hive_compatible_tbl (  
  2.     `user_id` BIGINT, -- 用户id 
  3.     `item_id` BIGINT, -- 商品id 
  4.     `cat_id` BIGINT, -- 品类id 
  5.     `action` STRING, -- 用户行为 
  6.     `province` INT, -- 用户所在的省份 
  7.     `ts` BIGINT -- 用户行为发生的时间戳 
  8.  ) WITH (  
  9.     'connector' = 'kafka', -- 使用 kafka connector 
  10.     'topic' = 'user_behavior', -- kafka主题 
  11.     'scan.startup.mode' = 'earliest-offset', -- 偏移量 
  12.     'properties.group.id' = 'group1', -- 消费者组 
  13.     'properties.bootstrap.servers' = 'kms-2:9092,kms-3:9092,kms-4:9092',  
  14.     'format' = 'json', -- 数据源格式为json 
  15.     'json.fail-on-missing-field' = 'true', 
  16.     'json.ignore-parse-errors' = 'false', 
  17.     'is_generic' = 'false' 
  18. ); 

当我们在Hive中查看该表的元数据信息时，可以看出：is_generic =false

  1. hive (default)> desc formatted hive_compatible_tbl; 
  2. Table Parameters:                 
  3.         ...            
  4.         is_generic              false                
  5.         ... 

我们可以使用FlinkSQL Cli或者HiveCli向该表中写入数据，然后分别通过FlinkSQL Cli和Hive Cli去查看该表数据的变化

  1. hive (default)> insert into hive_compatible_tbl select 2020,1221,100,'buy',11,1574330486; 
  2. hive (default)> select * from hive_compatible_tbl; 

再在FlinkSQL Cli中查看该表，

  1. Flink SQL> select user_id,item_id,action from hive_compatible_tbl; 
  2.                    user_id                   item_id                    action 
  3.                       2020                      1221                       buy 
  4.      

同样，我们可以在FlinkSQL Cli中去向该表中写入数据：

  1. Flink SQL>  insert into hive_compatible_tbl select 2020,1222,101,'fav',11,1574330486; 
  2. Flink SQL> select user_id,item_id,action from hive_compatible_tbl; 
  3.  
  4.                    user_id                   item_id                    action 
  5.                       2020                      1221                       buy 
  6.                       2020                      1222                       fav 

尖叫提示：

对于Hive兼容的表，需要注意数据类型，具体的数据类型对应关系以及注意点如下

Flink 数据类型 Hive 数据类型
CHAR(p) CHAR(p)
VARCHAR(p) VARCHAR(p)
STRING STRING
BOOLEAN BOOLEAN
TINYINT TINYINT
SMALLINT SMALLINT
INT INT
BIGINT LONG
FLOAT FLOAT
DOUBLE DOUBLE
DECIMAL(p, s) DECIMAL(p, s)
DATE DATE
TIMESTAMP(9) TIMESTAMP
BYTES BINARY
ARRAY LIST
MAP MAP
ROW STRUCT

注意：

什么是Hive Dialect

从Flink1.11.0开始，只要开启了Hive dialect配置，用户就可以使用HiveQL语法，这样我们就可以在Flink中使用Hive的语法使用一些DDL和DML操作。

Flink目前支持两种SQL方言(SQL dialects),分别为：default和hive。默认的SQL方言是default，如果要使用Hive的语法，需要将SQL方言切换到hive。

如何使用Hive Dialect

在SQL Cli中使用Hive dialect

使用hive dialect只需要配置一个参数即可，该参数名称为：table.sql-dialect。我们就可以在sql-client-defaults.yaml配置文件中进行配置，也可以在具体的会话窗口中进行设定，对于SQL dialect的切换，不需要进行重启session。

  1. execution: 
  2.   planner: blink 
  3.   type: batch 
  4.   result-mode: table 
  5.  
  6. configuration: 
  7.   table.sql-dialect: hive 

如果我们需要在SQL Cli中进行切换hive dialect，可以使用如下命令：

  1. Flink SQL> set table.sql-dialect=hive; -- 使用hive dialect 
  2. Flink SQL> set table.sql-dialect=default; -- 使用default dialect 

尖叫提示：

一旦切换到了hive dialect，就只能使用Hive的语法建表，如果尝试使用Flink的语法建表，则会报错

在Table API中配合dialect

  1. EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner()...build(); 
  2. TableEnvironment tableEnv = TableEnvironment.create(settings); 
  3. // 使用hive dialect 
  4. tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE); 
  5. // 使用 default dialect 
  6. tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT); 

操作示例

  1. Flink SQL> set table.sql-dialect=hive; 
  2. -- 使用Hive语法创建一张表 
  3. CREATE TABLE IF NOT EXISTS `hive_dialect_tbl` ( 
  4.   `id` int COMMENT 'id', 
  5.   `name` string COMMENT '名称', 
  6.   `age` int COMMENT '年龄'  
  7. ) 
  8. COMMENT 'hive dialect表测试' 
  9. ROW FORMAT DELIMITED FIELDS TERMINATED BY ','; 

进入Hive客户端去查看该表的元数据信息

  1. desc formatted hive_dialect_tbl; 
  2. col_name        data_type       comment 
  3. # col_name              data_type               comment              
  4.                   
  5. id                      int                                          
  6. name                    string                                       
  7. age                     int                                          
  8.                   
  9. # Detailed Table Information              
  10. Database:               default                   
  11. Owner:                  null                      
  12. CreateTime:             Mon Dec 21 17:23:48 CST 2020      
  13. LastAccessTime:         UNKNOWN                   
  14. Retention:              0                         
  15. Location:               hdfs://kms-1.apache.com:8020/user/hive/warehouse/hive_dialect_tbl         
  16. Table Type:             MANAGED_TABLE             
  17. Table Parameters:                 
  18.         comment                 hive dialect表测试      
  19.         is_generic              false                
  20.         transient_lastDdlTime   1608542628           
  21.                   
  22. # Storage Information             
  23. SerDe Library:          org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe        
  24. InputFormat:            org.apache.hadoop.mapred.TextInputFormat          
  25. OutputFormat:           org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat    
  26. Compressed:             No                        
  27. Num Buckets:            -1                        
  28. Bucket Columns:         []                        
  29. Sort Columns:           []                        
  30. Storage Desc Params:              
  31.         field.delim             ,                    
  32.         serialization.format    ,                    

很明显，该表是一张Hive兼容表，即is_generic=false。

使用FlinkSQLCli向该表中写入一条数据：

  1. Flink SQL> insert into hive_dialect_tbl select 1,'tom',20; 

我们也可以在Hive的Cli中去操作该表

  1. hive (default)> select * from hive_dialect_tbl; 
  2. hive (default)> insert into hive_dialect_tbl select 2,'jack',22; 

以下是使用Hive方言的一些注意事项。

当然，一旦开启了Hive dialect，我们就可以按照Hive的操作方式在Flink中去处理Hive的数据了，具体的操作与Hive一致，本文不再赘述。

Flink读写Hive

Flink写入Hive表

Flink支持以**批处理(Batch)和流处理(Streaming)**的方式写入Hive表。当以批处理的方式写入Hive表时，只有当写入作业结束时，才可以看到写入的数据。批处理的方式写入支持append模式和overwrite模式。

批处理模式写入

向非分区表写入数据

  1. Flink SQL> use catalog myhive; -- 使用catalog 
  2. Flink SQL> INSERT INTO users SELECT 2,'tom'; 
  3. Flink SQL> set execution.type=batch; -- 使用批处理模式 
  4. Flink SQL> INSERT OVERWRITE users SELECT 2,'tom'; 

向分区表写入数据

  1. -- 向静态分区表写入数据 
  2. Flink SQL> INSERT OVERWRITE myparttable PARTITION (my_type='type_1', my_date='2019-08-08') SELECT 'Tom', 25; 
  3. -- 向动态分区表写入数据 
  4. Flink SQL> INSERT OVERWRITE myparttable SELECT 'Tom', 25, 'type_1', '2019-08-08'; 

流处理模式写入

流式写入Hive表，不支持**Insert overwrite **方式，否则报如下错误：

  1. [ERROR] Could not execute SQL statement. Reason: 
  2. java.lang.IllegalStateException: Streaming mode not support overwrite. 

下面的示例是将kafka的数据流式写入Hive的分区表

  1. -- 使用流处理模式 
  2. Flink SQL> set execution.type=streaming; 
  3. -- 使用Hive方言 
  4. Flink SQL> SET table.sql-dialect=hive;  
  5. -- 创建一张Hive分区表 
  6. CREATE TABLE user_behavior_hive_tbl ( 
  7.    `user_id` BIGINT, -- 用户id 
  8.     `item_id` BIGINT, -- 商品id 
  9.     `cat_id` BIGINT, -- 品类id 
  10.     `action` STRING, -- 用户行为 
  11.     `province` INT, -- 用户所在的省份 
  12.     `ts` BIGINT -- 用户行为发生的时间戳 
  13. ) PARTITIONED BY (dt STRING,hr STRING,mi STRING) STORED AS parquet  TBLPROPERTIES ( 
  14.   'partition.time-extractor.timestamp-pattern'='$dt $hr:$mi:00', 
  15.   'sink.partition-commit.trigger'='partition-time', 
  16.   'sink.partition-commit.delay'='0S', 
  17.   'sink.partition-commit.policy.kind'='metastore,success-file' 
  18. ); 
  19.  
  20. -- 使用默认SQL方言 
  21. Flink SQL> SET table.sql-dialect=default;  
  22. -- 创建一张kafka数据源表 
  23. CREATE TABLE user_behavior (  
  24.     `user_id` BIGINT, -- 用户id 
  25.     `item_id` BIGINT, -- 商品id 
  26.     `cat_id` BIGINT, -- 品类id 
  27.     `action` STRING, -- 用户行为 
  28.     `province` INT, -- 用户所在的省份 
  29.     `ts` BIGINT, -- 用户行为发生的时间戳 
  30.     `proctime` AS PROCTIME(), -- 通过计算列产生一个处理时间列 
  31.     `eventTime` AS TO_TIMESTAMP(FROM_UNIXTIME(ts, 'yyyy-MM-dd HH:mm:ss')), -- 事件时间 
  32.      WATERMARK FOR eventTime AS eventTime - INTERVAL '5' SECOND  -- 定义watermark 
  33.  ) WITH (  
  34.     'connector' = 'kafka', -- 使用 kafka connector 
  35.     'topic' = 'user_behaviors', -- kafka主题 
  36.     'scan.startup.mode' = 'earliest-offset', -- 偏移量 
  37.     'properties.group.id' = 'group1', -- 消费者组 
  38.     'properties.bootstrap.servers' = 'kms-2:9092,kms-3:9092,kms-4:9092',  
  39.     'format' = 'json', -- 数据源格式为json 
  40.     'json.fail-on-missing-field' = 'true', 
  41.     'json.ignore-parse-errors' = 'false' 
  42. ); 

关于Hive表的一些属性解释：

如果是按天时进行分区，则该属性值为：dt $hour:00:00`;

sink.partition-commit.delay

sink.partition-commit.policy.kind

执行流式写入Hive表

  1. -- streaming sql,将数据写入Hive表 
  2. INSERT INTO user_behavior_hive_tbl  
  3. SELECT  
  4.     user_id, 
  5.     item_id, 
  6.     cat_id, 
  7.     action, 
  8.     province, 
  9.     ts, 
  10.     FROM_UNIXTIME(ts, 'yyyy-MM-dd'), 
  11.     FROM_UNIXTIME(ts, 'HH'), 
  12.     FROM_UNIXTIME(ts, 'mm') 
  13. FROM user_behavior; 
  14.  
  15. -- batch sql,查询Hive表的分区数据 
  16. SELECT * FROM user_behavior_hive_tbl WHERE dt='2021-01-04' AND  hr='16' AND mi = '46'; 

同时查看Hive表的分区数据：

尖叫提示：

Flink读取Hive表默认使用的是batch模式，如果要使用流式读取Hive表，需要而外指定一些参数，见下文。

只有在完成 Checkpoint 之后，文件才会从 In-progress 状态变成 Finish 状态，同时生成_SUCCESS文件，所以，Flink流式写入Hive表需要开启并配置 Checkpoint。对于Flink SQL Client而言，需要在flink-conf.yaml中开启CheckPoint，配置内容为：

state.backend: filesystem execution.checkpointing.externalized-checkpoint-retention:RETAIN_ON_CANCELLATION execution.checkpointing.interval: 60s execution.checkpointing.mode: EXACTLY_ONCE state.savepoints.dir: hdfs://kms-1:8020/flink-savepoints

Flink读取Hive表

Flink支持以**批处理(Batch)和流处理(Streaming)**的方式读取Hive中的表。批处理的方式与Hive的本身查询类似，即只在提交查询的时刻查询一次Hive表。流处理的方式将会持续地监控Hive表，并且会增量地提取新的数据。默认情况下，Flink是以批处理的方式读取Hive表。

关于流式读取Hive表，Flink既支持分区表又支持非分区表。对于分区表而言，Flink将会监控新产生的分区数据，并以增量的方式读取这些数据。对于非分区表，Flink会监控Hive表存储路径文件夹里面的新文件，并以增量的方式读取新的数据。

Flink读取Hive表可以配置一下参数：

在 SQL Client 中需要显示地开启 SQL Hint 功能

  1. Flink SQL> set table.dynamic-table-options.enabled= true;   

使用SQLHint流式查询Hive表

  1. SELECT * FROM user_behavior_hive_tbl ; 

Hive维表JOIN

Flink 1.12 支持了 Hive 最新的分区作为时态表的功能，可以通过 SQL 的方式直接关联 Hive 分区表的最新分区，并且会自动监听最新的 Hive 分区，当监控到新的分区后，会自动地做维表数据的全量替换。

Flink支持的是processing-time的temporal join，也就是说总是与最新版本的时态表进行JOIN。另外，Flink既支持非分区表的temporal join，又支持分区表的temporal join。对于分区表而言，Flink会监听Hive表的最新分区数据。值得注意的是，Flink尚不支持 event-time temporal join。

Temporal Join最新分区

对于一张随着时间变化的Hive分区表，Flink可以读取该表的数据作为一个无界流。如果Hive分区表的每个分区都包含全量的数据，那么每个分区将做为一个时态表的版本数据，即将最新的分区数据作为一个全量维表数据。值得注意的是，该功能特点仅支持Flink的STREAMING模式。

使用 Hive 最新分区作为 Tempmoral table 之前，需要设置必要的两个参数：

  1. 'streaming-source.enable' = 'true',   
  2. 'streaming-source.partition.include' = 'latest' 

除此之外还有一些其他的参数，关于参数的解释见上面的分析。我们在使用Hive维表的时候，既可以在创建Hive表时指定具体的参数，也可以使用SQL Hint的方式动态指定参数。一个Hive维表的创建模板如下：

  1. -- 使用Hive的sql方言 
  2. SET table.sql-dialect=hive; 
  3. CREATE TABLE dimension_table ( 
  4.   product_id STRING, 
  5.   product_name STRING, 
  6.   unit_price DECIMAL(10, 4), 
  7.   pv_count BIGINT, 
  8.   like_count BIGINT, 
  9.   comment_count BIGINT, 
  10.   update_time TIMESTAMP(3), 
  11.   update_user STRING, 
  12.   ... 
  13. ) PARTITIONED BY (pt_year STRING, pt_month STRING, pt_day STRING) TBLPROPERTIES ( 
  14.   -- 方式1：按照分区名排序来识别最新分区(推荐使用该种方式) 
  15.   'streaming-source.enable' = 'true', -- 开启Streaming source 
  16.   'streaming-source.partition.include' = 'latest',-- 选择最新分区 
  17.   'streaming-source.monitor-interval' = '12 h',-- 每12小时加载一次最新分区数据 
  18.   'streaming-source.partition-order' = 'partition-name',  -- 按照分区名排序 
  19.  
  20.   -- 方式2:分区文件的创建时间排序来识别最新分区 
  21.   'streaming-source.enable' = 'true', 
  22.   'streaming-source.partition.include' = 'latest', 
  23.   'streaming-source.partition-order' = 'create-time',-- 分区文件的创建时间排序 
  24.   'streaming-source.monitor-interval' = '12 h' 
  25.  
  26.   -- 方式3:按照分区时间排序来识别最新分区 
  27.   'streaming-source.enable' = 'true', 
  28.   'streaming-source.partition.include' = 'latest', 
  29.   'streaming-source.monitor-interval' = '12 h', 
  30.   'streaming-source.partition-order' = 'partition-time', -- 按照分区时间排序 
  31.   'partition.time-extractor.kind' = 'default', 
  32.   'partition.time-extractor.timestamp-pattern' = '$pt_year-$pt_month-$pt_day 00:00:00'  
  33. ); 

有了上面的Hive维表，我们就可以使用该维表与Kafka的实时流数据进行JOIN，得到相应的宽表数据。

  1. -- 使用default sql方言 
  2. SET table.sql-dialect=default; 
  3. -- kafka实时流数据表 
  4. CREATE TABLE orders_table ( 
  5.   order_id STRING, 
  6.   order_amount DOUBLE, 
  7.   product_id STRING, 
  8.   log_ts TIMESTAMP(3), 
  9.   proctime as PROCTIME() 
  10. ) WITH (...); 
  11.  
  12. -- 将流表与hive最新分区数据关联  
  13. SELECT * 
  14. FROM orders_table AS orders 
  15. JOIN dimension_table FOR SYSTEM_TIME AS OF orders.proctime AS dim  
  16. ON orders.product_id = dim.product_id; 

除了在定义Hive维表时指定相关的参数，我们还可以通过SQL Hint的方式动态指定相关的参数，具体方式如下：

  1. SELECT * 
  2. FROM orders_table AS orders 
  3. JOIN dimension_table 
  4.  
  5. FOR SYSTEM_TIME AS OF orders.proctime AS dim -- 时态表(维表) 
  6. ON orders.product_id = dim.product_id; 

Temporal Join最新表

对于Hive的非分区表，当使用temporal join时，整个Hive表会被缓存到Slot内存中，然后根据流中的数据对应的key与其进行匹配。使用最新的Hive表进行temporal join不需要进行额外的配置，我们只需要配置一个Hive表缓存的TTL时间，该时间的作用是：当缓存过期时，就会重新扫描Hive表并加载最新的数据。

尖叫提示：

当使用此种方式时，Hive表必须是有界的lookup表，即非Streaming Source的时态表，换句话说，该表的属性streaming-source.enable = false。

如果要使用Streaming Source的时态表，记得配置streaming-source.monitor-interval的值，即数据更新的时间间隔。

  1. -- Hive维表数据使用批处理的方式按天装载 
  2. SET table.sql-dialect=hive; 
  3. CREATE TABLE dimension_table ( 
  4.   product_id STRING, 
  5.   product_name STRING, 
  6.   unit_price DECIMAL(10, 4), 
  7.   pv_count BIGINT, 
  8.   like_count BIGINT, 
  9.   comment_count BIGINT, 
  10.   update_time TIMESTAMP(3), 
  11.   update_user STRING, 
  12.   ... 
  13. ) TBLPROPERTIES ( 
  14.   'streaming-source.enable' = 'false', -- 关闭streaming source 
  15.   'streaming-source.partition.include' = 'all',  -- 读取所有数据 
  16.   'lookup.join.cache.ttl' = '12 h' 
  17. ); 
  18. -- kafka事实表 
  19. SET table.sql-dialect=default; 
  20. CREATE TABLE orders_table ( 
  21.   order_id STRING, 
  22.   order_amount DOUBLE, 
  23.   product_id STRING, 
  24.   log_ts TIMESTAMP(3), 
  25.   proctime as PROCTIME() 
  26. ) WITH (...); 
  27.  
  28. -- Hive维表join，Flink会加载该维表的所有数据到内存中 
  29. SELECT * 
  30. FROM orders_table AS orders 
  31. JOIN dimension_table FOR SYSTEM_TIME AS OF orders.proctime AS dim 
  32. ON orders.product_id = dim.product_id; 

尖叫提示：

每一个子任务都需要缓存一份维表的全量数据，一定要确保TM的task Slot 大小能够容纳维表的数据量;

推荐将streaming-source.monitor-interval和lookup.join.cache.ttl的值设为一个较大的数，因为频繁的更新和加载数据会影响性能。

当缓存的维表数据需要重新刷新时，目前的做法是将整个表进行加载，因此不能够将新数据与旧数据区分开来。

Hive维表JOIN示例

假设维表的数据是通过批处理的方式(比如每天)装载至Hive中，而Kafka中的事实流数据需要与该维表进行JOIN，从而构建一个宽表数据，这个时候就可以使用Hive的维表JOIN。

  1. SET table.sql-dialect=default; 
  2. CREATE TABLE fact_user_behavior (  
  3.     `user_id` BIGINT, -- 用户id 
  4.     `item_id` BIGINT, -- 商品id 
  5.     `action` STRING, -- 用户行为 
  6.     `province` INT, -- 用户所在的省份 
  7.     `ts` BIGINT, -- 用户行为发生的时间戳 
  8.     `proctime` AS PROCTIME(), -- 通过计算列产生一个处理时间列 
  9.     `eventTime` AS TO_TIMESTAMP(FROM_UNIXTIME(ts, 'yyyy-MM-dd HH:mm:ss')), -- 事件时间 
  10.      WATERMARK FOR eventTime AS eventTime - INTERVAL '5' SECOND  -- 定义watermark 
  11.  ) WITH (  
  12.     'connector' = 'kafka', -- 使用 kafka connector 
  13.     'topic' = 'user_behaviors', -- kafka主题 
  14.     'scan.startup.mode' = 'earliest-offset', -- 偏移量 
  15.     'properties.group.id' = 'group1', -- 消费者组 
  16.     'properties.bootstrap.servers' = 'kms-2:9092,kms-3:9092,kms-4:9092',  
  17.     'format' = 'json', -- 数据源格式为json 
  18.     'json.fail-on-missing-field' = 'true', 
  19.     'json.ignore-parse-errors' = 'false' 
  20. ); 
  1. SET table.sql-dialect=hive; 
  2. CREATE TABLE dim_item ( 
  3.   item_id BIGINT, 
  4.   item_name STRING, 
  5.   unit_price DECIMAL(10, 4) 
  6. ) PARTITIONED BY (dt STRING) TBLPROPERTIES ( 
  7.   'streaming-source.enable' = 'true', 
  8.   'streaming-source.partition.include' = 'latest', 
  9.   'streaming-source.monitor-interval' = '12 h', 
  10.   'streaming-source.partition-order' = 'partition-name' 
  11. ); 
  1. SELECT  
  2.     fact.item_id, 
  3.     dim.item_name, 
  4.     count(*) AS buy_cnt 
  5. FROM fact_user_behavior AS fact 
  6. LEFT JOIN dim_item FOR SYSTEM_TIME AS OF fact.proctime AS dim 
  7. ON fact.item_id = dim.item_id 
  8. WHERE fact.action = 'buy' 
  9. GROUP BY fact.item_id,dim.item_name; 

使用SQL Hint方式，关联非分区的Hive维表：

  1. set table.dynamic-table-options.enabled= true;  
  2. SELECT  
  3.     fact.item_id, 
  4.     dim.item_name, 
  5.     count(*) AS buy_cnt 
  6. FROM fact_user_behavior AS fact 
  7. LEFT JOIN dim_item1 
  8.  
  9. FOR SYSTEM_TIME AS OF fact.proctime AS dim 
  10. ON fact.item_id = dim.item_id 
  11. WHERE fact.action = 'buy' 
  12. GROUP BY fact.item_id,dim.item_name; 

Flink upsert-kafka连接器

Upsert Kafka connector简介

Upsert Kafka Connector允许用户以upsert的方式从Kafka主题读取数据或将数据写入Kafka主题。

当作为数据源时，upsert-kafka Connector会生产一个changelog流，其中每条数据记录都表示一个更新或删除事件。更准确地说，如果不存在对应的key，则视为INSERT操作。如果已经存在了相对应的key，则该key对应的value值为最后一次更新的值。

用表来类比，changelog 流中的数据记录被解释为 UPSERT，也称为 INSERT/UPDATE，因为任何具有相同 key 的现有行都被覆盖。另外，value 为空的消息将会被视作为 DELETE 消息。

当作为数据汇时，upsert-kafka Connector会消费一个changelog流。它将INSERT / UPDATE_AFTER数据作为正常的Kafka消息值写入(即INSERT和UPDATE操作，都会进行正常写入，如果是更新，则同一个key会存储多条数据，但在读取该表数据时，只保留最后一次更新的值)，并将 DELETE 数据以 value 为空的 Kafka 消息写入(key被打上墓碑标记，表示对应 key 的消息被删除)。Flink 将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新/删除消息将落在同一分区中

依赖

为了使用Upsert Kafka连接器，需要添加下面的依赖

  1.  
  2.     org.apache.flink 
  3.     flink-connector-kafka_2.12 
  4.     1.12.0 
  5.  

如果使用SQL Client，需要下载flink-sql-connector-kafka_2.11-1.12.0.jar，并将其放置在Flink安装目录的lib文件夹下。

使用方式

使用样例

  1. -- 创建一张kafka表，用户存储sink的数据 
  2. CREATE TABLE pageviews_per_region ( 
  3.   user_region STRING, 
  4.   pv BIGINT, 
  5.   uv BIGINT, 
  6.   PRIMARY KEY (user_region) NOT ENFORCED 
  7. ) WITH ( 
  8.   'connector' = 'upsert-kafka', 
  9.   'topic' = 'pageviews_per_region', 
  10.   'properties.bootstrap.servers' = 'kms-2:9092,kms-3:9092,kms-4:9092', 
  11.   'key.format' = 'avro', 
  12.   'value.format' = 'avro' 
  13. ); 

尖叫提示：

要使用 upsert-kafka connector，必须在创建表时使用PRIMARY KEY定义主键，并为键(key.format)和值(value.format)指定序列化反序列化格式。

upsert-kafka connector参数

必选。指定要使用的连接器，Upsert Kafka 连接器使用：'upsert-kafka'。

必选。用于读取和写入的 Kafka topic 名称。

必选。以逗号分隔的 Kafka brokers 列表。

必选。用于对 Kafka 消息中 key 部分序列化和反序列化的格式。key 字段由 PRIMARY KEY 语法指定。支持的格式包括 'csv'、'json'、'avro'。

必选。用于对 Kafka 消息中 value 部分序列化和反序列化的格式。支持的格式包括 'csv'、'json'、'avro'。

可选。该选项可以传递任意的 Kafka 参数。选项的后缀名必须匹配定义在 Kafka 参数文档中的参数名。Flink 会自动移除 选项名中的 "properties." 前缀，并将转换后的键名以及值传入 KafkaClient。例如，你可以通过 'properties.allow.auto.create.topics' = 'false' 来禁止自动创建 topic。但是，某些选项，例如'key.deserializer' 和 'value.deserializer' 是不允许通过该方式传递参数，因为 Flink 会重写这些参数的值。

可选，默认为ALL。控制key字段是否出现在 value 中。当取ALL时，表示消息的 value 部分将包含 schema 中所有的字段，包括定义为主键的字段。当取EXCEPT_KEY时，表示记录的 value 部分包含 schema 的所有字段，定义为主键的字段除外。

可选。为了避免与value字段命名冲突，为key字段添加一个自定义前缀。默认前缀为空。一旦指定了key字段的前缀，必须在DDL中指明前缀的名称，但是在构建key的序列化数据类型时，将移除该前缀。见下面的示例。在需要注意的是：使用该配置属性，value.fields-include的值必须为EXCEPT_KEY。

  1. -- 创建一张upsert表，当指定了qwe前缀，涉及的key必须指定qwe前缀 
  2. CREATE TABLE result_total_pvuv_min_prefix ( 
  3.     qwedo_date     STRING,     -- 统计日期，必须包含qwe前缀 
  4.     qwedo_min      STRING,      -- 统计分钟，必须包含qwe前缀 
  5.     pv          BIGINT,     -- 点击量 
  6.     uv          BIGINT,     -- 一天内同个访客多次访问仅计算一个UV 
  7.     currenttime TIMESTAMP,  -- 当前时间 
  8.     PRIMARY KEY (qwedo_date, qwedo_min) NOT ENFORCED -- 必须包含qwe前缀 
  9. ) WITH ( 
  10.   'connector' = 'upsert-kafka', 
  11.   'topic' = 'result_total_pvuv_min_prefix', 
  12.   'properties.bootstrap.servers' = 'kms-2:9092,kms-3:9092,kms-4:9092', 
  13.   'key.json.ignore-parse-errors' = 'true', 
  14.   'value.json.fail-on-missing-field' = 'false', 
  15.   'key.format' = 'json', 
  16.   'value.format' = 'json', 
  17.   'key.fields-prefix'='qwe', -- 指定前缀qwe 
  18.   'value.fields-include' = 'EXCEPT_KEY' -- key不出现kafka消息的value中 
  19. ); 
  20. -- 向该表中写入数据 
  21. INSERT INTO result_total_pvuv_min_prefix 
  22. SELECT 
  23.   do_date,    --  时间分区 
  24.   cast(DATE_FORMAT (access_time,'HH:mm') AS STRING) AS do_min,-- 分钟级别的时间 
  25.   pv, 
  26.   uv, 
  27.   CURRENT_TIMESTAMP AS currenttime -- 当前时间 
  28. from 
  29.   view_total_pvuv_min; 

尖叫提示：

如果指定了key字段前缀，但在DDL中并没有添加该前缀字符串，那么在向该表写入数时，会抛出下面异常：

[ERROR] Could not execute SQL statement. Reason: org.apache.flink.table.api.ValidationException: All fields in 'key.fields' must be prefixed with 'qwe' when option 'key.fields-prefix' is set but field 'do_date' is not prefixed.

可选。定义 upsert-kafka sink 算子的并行度。默认情况下，由框架确定并行度，与上游链接算子的并行度保持一致。

其他注意事项

Key和Value的序列化格式

关于Key、value的序列化可以参考Kafka connector。值得注意的是，必须指定Key和Value的序列化格式，其中Key是通过PRIMARY KEY指定的。

Primary Key约束

Upsert Kafka 工作在 upsert 模式(FLIP-149)下。当我们创建表时，需要在 DDL 中定义主键。具有相同key的数据，会存在相同的分区中。在 changlog source 上定义主键意味着在物化后的 changelog 上主键具有唯一性。定义的主键将决定哪些字段出现在 Kafka 消息的 key 中。

一致性保障

默认情况下，如果启用 checkpoint，Upsert Kafka sink 会保证至少一次将数据插入 Kafka topic。

这意味着，Flink 可以将具有相同 key 的重复记录写入 Kafka topic。但由于该连接器以 upsert 的模式工作，该连接器作为 source 读入时，可以确保具有相同主键值下仅最后一条消息会生效。因此，upsert-kafka 连接器可以像 HBase sink 一样实现幂等写入。

分区水位线

Flink 支持根据 Upsert Kafka 的 每个分区的数据特性发送相应的 watermark。当使用这个特性的时候，watermark 是在 Kafka consumer 内部生成的。合并每个分区生成的 watermark 的方式和 streaming shuffle 的方式是一致的(单个分区的输入取最大值，多个分区的输入取最小值)。数据源产生的 watermark 是取决于该 consumer 负责的所有分区中当前最小的 watermark。如果该 consumer 负责的部分分区是空闲的，那么整体的 watermark 并不会前进。在这种情况下，可以通过设置合适的 table.exec.source.idle-timeout 来缓解这个问题。

数据类型

Upsert Kafka 用字节bytes存储消息的 key 和 value，因此没有 schema 或数据类型。消息按格式进行序列化和反序列化，例如：csv、json、avro。不同的序列化格式所提供的数据类型有所不同，因此需要根据使用的序列化格式进行确定表字段的数据类型是否与该序列化类型提供的数据类型兼容。

使用案例

本文以实时地统计网页PV和UV的总量为例，介绍upsert-kafka基本使用方式：

用户的ippv信息，一个用户在一天内可以有很多次pv

  1. CREATE TABLE source_ods_fact_user_ippv ( 
  2.     user_id      STRING,       -- 用户ID 
  3.     client_ip    STRING,       -- 客户端IP 
  4.     client_info  STRING,       -- 设备机型信息 
  5.     pagecode     STRING,       -- 页面代码 
  6.     access_time  TIMESTAMP,    -- 请求时间 
  7.     dt           STRING,       -- 时间分区天 
  8.     WATERMARK FOR access_time AS access_time - INTERVAL '5' SECOND  -- 定义watermark 
  9. ) WITH ( 
  10.    'connector' = 'kafka', -- 使用 kafka connector 
  11.     'topic' = 'user_ippv', -- kafka主题 
  12.     'scan.startup.mode' = 'earliest-offset', -- 偏移量 
  13.     'properties.group.id' = 'group1', -- 消费者组 
  14.     'properties.bootstrap.servers' = 'kms-2:9092,kms-3:9092,kms-4:9092',  
  15.     'format' = 'json', -- 数据源格式为json 
  16.     'json.fail-on-missing-field' = 'false', 
  17.     'json.ignore-parse-errors' = 'true' 
  18. ); 

统计每分钟的PV、UV，并将结果存储在Kafka中

  1. CREATE TABLE result_total_pvuv_min ( 
  2.     do_date     STRING,     -- 统计日期 
  3.     do_min      STRING,      -- 统计分钟 
  4.     pv          BIGINT,     -- 点击量 
  5.     uv          BIGINT,     -- 一天内同个访客多次访问仅计算一个UV 
  6.     currenttime TIMESTAMP,  -- 当前时间 
  7.     PRIMARY KEY (do_date, do_min) NOT ENFORCED 
  8. ) WITH ( 
  9.   'connector' = 'upsert-kafka', 
  10.   'topic' = 'result_total_pvuv_min', 
  11.   'properties.bootstrap.servers' = 'kms-2:9092,kms-3:9092,kms-4:9092', 
  12.   'key.json.ignore-parse-errors' = 'true', 
  13.   'value.json.fail-on-missing-field' = 'false', 
  14.   'key.format' = 'json', 
  15.   'value.format' = 'json', 
  16.   'value.fields-include' = 'EXCEPT_KEY' -- key不出现kafka消息的value中 
  17. ); 
  1. -- 创建视图 
  2. CREATE VIEW view_total_pvuv_min AS 
  3. SELECT 
  4.      dt AS do_date,                    -- 时间分区 
  5.      count (client_ip) AS pv,          -- 客户端的IP 
  6.      count (DISTINCT client_ip) AS uv, -- 客户端去重 
  7.      max(access_time) AS access_time   -- 请求的时间 
  8. FROM 
  9.     source_ods_fact_user_ippv 
  10. GROUP BY dt; 
  11.  
  12. -- 写入数据 
  13. INSERT INTO result_total_pvuv_min 
  14. SELECT 
  15.   do_date,    --  时间分区 
  16.   cast(DATE_FORMAT (access_time,'HH:mm') AS STRING) AS do_min,-- 分钟级别的时间 
  17.   pv, 
  18.   uv, 
  19.   CURRENT_TIMESTAMP AS currenttime -- 当前时间 
  20. from 
  21.   view_total_pvuv_min; 
  1. {"user_id":"1","client_ip":"192.168.12.1","client_info":"phone","pagecode":"1001","access_time":"2021-01-08 11:32:24","dt":"2021-01-08"} 
  2.  
  3. {"user_id":"1","client_ip":"192.168.12.1","client_info":"phone","pagecode":"1201","access_time":"2021-01-08 11:32:55","dt":"2021-01-08"} 
  4.  
  5. {"user_id":"2","client_ip":"192.165.12.1","client_info":"pc","pagecode":"1031","access_time":"2021-01-08 11:32:59","dt":"2021-01-08"} 
  6.  
  7. {"user_id":"1","client_ip":"192.168.12.1","client_info":"phone","pagecode":"1101","access_time":"2021-01-08 11:33:24","dt":"2021-01-08"} 
  8.  
  9. {"user_id":"3","client_ip":"192.168.10.3","client_info":"pc","pagecode":"1001","access_time":"2021-01-08 11:33:30","dt":"2021-01-08"} 
  10.  
  11. {"user_id":"1","client_ip":"192.168.12.1","client_info":"phone","pagecode":"1001","access_time":"2021-01-08 11:34:24","dt":"2021-01-08"} 
  1. select * from result_total_pvuv_min; 

可以看出：每分钟的pv、uv只显示一条数据，即代表着截止到当前时间点的pv和uv

查看Kafka中result_total_pvuv_min主题的数据，如下：

可以看出：针对每一条访问数据，触发计算了一次PV、UV，每一条数据都是截止到当前时间的累计PV和UV。

尖叫提示：

默认情况下，如果在启用了检查点的情况下执行查询，Upsert Kafka接收器会将具有至少一次保证的数据提取到Kafka主题中。

这意味着，Flink可能会将具有相同键的重复记录写入Kafka主题。但是，由于连接器在upsert模式下工作，因此作为源读回时，同一键上的最后一条记录将生效。因此，upsert-kafka连接器就像HBase接收器一样实现幂等写入。

Flink CDC的connector

简介

Flink CDC Connector 是ApacheFlink的一组数据源连接器，使用变化数据捕获change data capture (CDC)从不同的数据库中提取变更数据。Flink CDC连接器将Debezium集成为引擎来捕获数据变更。因此，它可以充分利用Debezium的功能。

特点

使用场景

Flink提供的 table format

Flink提供了一系列可以用于table connector的table format，具体如下：

Formats Supported Connectors
CSV Apache Kafka, Filesystem
JSON Apache Kafka, Filesystem, Elasticsearch
Apache Avro Apache Kafka, Filesystem
Debezium CDC Apache Kafka
Canal CDC Apache Kafka
Apache Parquet Filesystem
Apache ORC Filesystem

使用过程中的注意点

使用MySQL CDC的注意点

如果要使用MySQL CDC connector，对于程序而言，需要添加如下依赖:

  1.  
  2.   com.alibaba.ververica 
  3.   flink-connector-mysql-cdc 
  4.   1.0.0 
  5.  

如果要使用Flink SQL Client，需要添加如下jar包：flink-sql-connector-mysql-cdc-1.0.0.jar，将该jar包放在Flink安装目录的lib文件夹下即可。

使用canal-json的注意点

如果要使用Kafka的canal-json，对于程序而言，需要添加如下依赖:

  1. -- universal --> 
  2.  
  3.     org.apache.flink 
  4.     flink-connector-kafka_2.11 
  5.     1.11.0 
  6.  

如果要使用Flink SQL Client，需要添加如下jar包：flink-sql-connector-kafka_2.11-1.11.0.jar，将该jar包放在Flink安装目录的lib文件夹下即可。由于Flink1.11的安装包 的lib目录下并没有提供该jar包，所以必须要手动添加依赖包，否则会报如下错误：

  1. [ERROR] Could not execute SQL statement. Reason: 
  2. org.apache.flink.table.api.ValidationException: Could not find any factory for identifier 'kafka' that implements 'org.apache.flink.table.factories.DynamicTableSourceFactory' in the classpath. 
  3.  
  4. Available factory identifiers are: 
  5.  
  6. datagen 
  7. mysql-cdc 

使用changelog-json的注意点

如果要使用Kafka的changelog-json Format，对于程序而言，需要添加如下依赖:

  1.  
  2.   com.alibaba.ververica 
  3.   flink-format-changelog-json 
  4.   1.0.0 
  5.  

如果要使用Flink SQL Client，需要添加如下jar包：flink-format-changelog-json-1.0.0.jar，将该jar包放在Flink安装目录的lib文件夹下即可。

mysql-cdc的操作实践

  1. -- MySQL 
  2.  
  3. DROP TABLE IF EXISTS `order_info`; 
  4. CREATE TABLE `order_info` ( 
  5.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '编号', 
  6.   `consignee` varchar(100) DEFAULT NULL COMMENT '收货人', 
  7.   `consignee_tel` varchar(20) DEFAULT NULL COMMENT '收件人电话', 
  8.   `total_amount` decimal(10,2) DEFAULT NULL COMMENT '总金额', 
  9.   `order_status` varchar(20) DEFAULT NULL COMMENT '订单状态,1表示下单，2表示支付', 
  10.   `user_id` bigint(20) DEFAULT NULL COMMENT '用户id', 
  11.   `payment_way` varchar(20) DEFAULT NULL COMMENT '付款方式', 
  12.   `delivery_address` varchar(1000) DEFAULT NULL COMMENT '送货地址', 
  13.   `order_comment` varchar(200) DEFAULT NULL COMMENT '订单备注', 
  14.   `out_trade_no` varchar(50) DEFAULT NULL COMMENT '订单交易编号（第三方支付用)', 
  15.   `trade_body` varchar(200) DEFAULT NULL COMMENT '订单描述(第三方支付用)', 
  16.   `create_time` datetime DEFAULT NULL COMMENT '创建时间', 
  17.   `operate_time` datetime DEFAULT NULL COMMENT '操作时间', 
  18.   `expire_time` datetime DEFAULT NULL COMMENT '失效时间', 
  19.   `tracking_no` varchar(100) DEFAULT NULL COMMENT '物流单编号', 
  20.   `parent_order_id` bigint(20) DEFAULT NULL COMMENT '父订单编号', 
  21.   `img_url` varchar(200) DEFAULT NULL COMMENT '图片路径', 
  22.   `province_id` int(20) DEFAULT NULL COMMENT '地区', 
  23.   PRIMARY KEY (`id`) 
  24. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='订单表'; 
  25. -- ---------------------------- 
  26. -- Records of order_info 
  27. -- ---------------------------- 
  28. INSERT INTO `order_info`  
  29. VALUES (476, 'lAXjcL', '13408115089', 433.00, '2', 10, '2', 'OYyAdSdLxedceqovndCD', 'ihjAYsSjrgJMQVdFQnSy', '8728720206', '', '2020-06-18 02:21:38', NULL, NULL, NULL, NULL, NULL, 9); 
  30. INSERT INTO `order_info` 
  31. VALUES (477, 'QLiFDb', '13415139984', 772.00, '1', 90, '2', 'OizYrQbKuWvrvdfpkeSZ', 'wiBhhqhMndCCgXwmWVQq', '1679381473', '', '2020-06-18 09:12:25', NULL, NULL, NULL, NULL, NULL, 3); 
  32. INSERT INTO `order_info` 
  33. VALUES (478, 'iwKjQD', '13320383859', 88.00, '1', 107, '1', 'cbXLKtNHWOcWzJVBWdAs', 'njjsnknHxsxhuCCeNDDi', '0937074290', '', '2020-06-18 15:56:34', NULL, NULL, NULL, NULL, NULL, 7); 
  34.  
  35.  
  36. CREATE TABLE `order_detail` ( 
  37.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '编号', 
  38.   `order_id` bigint(20) DEFAULT NULL COMMENT '订单编号', 
  39.   `sku_id` bigint(20) DEFAULT NULL COMMENT 'sku_id', 
  40.   `sku_name` varchar(200) DEFAULT NULL COMMENT 'sku名称（冗余)', 
  41.   `img_url` varchar(200) DEFAULT NULL COMMENT '图片名称（冗余)', 
  42.   `order_price` decimal(10,2) DEFAULT NULL COMMENT '购买价格(下单时sku价格）', 
  43.   `sku_num` varchar(200) DEFAULT NULL COMMENT '购买个数', 
  44.   `create_time` datetime DEFAULT NULL COMMENT '创建时间', 
  45.   PRIMARY KEY (`id`) 
  46. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='订单明细表'; 
  47.  
  48. -- ---------------------------- 
  49. -- Records of order_detail 
  50. -- ---------------------------- 
  51. INSERT INTO `order_detail`  
  52. VALUES (1329, 476, 8, 'Apple iPhone XS Max (A2104) 256GB 深空灰色 移动联通电信4G手机 双卡双待', 'http://XLMByOyZDTJQYxphQHNTgYAFzJJCKTmCbzvEJIpz', 8900.00, '3', '2020-06-18 02:21:38'); 
  53. INSERT INTO `order_detail`  
  54. VALUES (1330, 477, 9, '荣耀10 GT游戏加速 AIS手持夜景 6GB+64GB 幻影蓝全网通 移动联通电信', 'http://ixOCtlYmlxEEgUfPLiLdjMftzrleOEIBKSjrhMne', 2452.00, '4', '2020-06-18 09:12:25'); 
  55. INSERT INTO `order_detail` 
  56. VALUES (1331, 478, 4, '小米Play 流光渐变AI双摄 4GB+64GB 梦幻蓝 全网通4G 双卡双待 小水滴全面屏拍照游戏智能手机', 'http://RqfEFnAOqnqRnNZLFRvBuwXxwNBtptYJCILDKQYv', 1442.00, '1', '2020-06-18 15:56:34'); 
  57. INSERT INTO `order_detail`  
  58. VALUES (1332, 478, 8, 'Apple iPhone XS Max (A2104) 256GB 深空灰色 移动联通电信4G手机 双卡双待', 'http://IwhuCDlsiLenfKjPzbJrIoxswdfofKhJLMzlJAKV', 8900.00, '3', '2020-06-18 15:56:34'); 
  59. INSERT INTO `order_detail`  
  60. VALUES (1333, 478, 8, 'Apple iPhone XS Max (A2104) 256GB 深空灰色 移动联通电信4G手机 双卡双待', 'http://bbfwTbAzTWapywODzOtDJMJUEqNTeRTUQuCDkqXP', 8900.00, '1', '2020-06-18 15:56:34'); 

Flink SQL Cli创建CDC数据源

启动 Flink 集群，再启动 SQL CLI,执行下面命令：

  1. -- 创建订单信息表 
  2. CREATE TABLE order_info( 
  3.     id BIGINT, 
  4.     user_id BIGINT, 
  5.     create_time TIMESTAMP(0), 
  6.     operate_time TIMESTAMP(0), 
  7.     province_id INT, 
  8.     order_status STRING, 
  9.     total_amount DECIMAL(10, 5) 
  10.   ) WITH ( 
  11.     'connector' = 'mysql-cdc', 
  12.     'hostname' = 'kms-1', 
  13.     'port' = '3306', 
  14.     'username' = 'root', 
  15.     'password' = '123qwe', 
  16.     'database-name' = 'mydw', 
  17.     'table-name' = 'order_info' 
  18. ); 

在Flink SQL Cli中查询该表的数据：result-mode: tableau，+表示数据的insert

在SQL CLI中创建订单详情表：

  1. CREATE TABLE order_detail( 
  2.     id BIGINT, 
  3.     order_id BIGINT, 
  4.     sku_id BIGINT, 
  5.     sku_name STRING, 
  6.     sku_num BIGINT, 
  7.     order_price DECIMAL(10, 5), 
  8.  create_time TIMESTAMP(0) 
  9.  ) WITH ( 
  10.     'connector' = 'mysql-cdc', 
  11.     'hostname' = 'kms-1', 
  12.     'port' = '3306', 
  13.     'username' = 'root', 
  14.     'password' = '123qwe', 
  15.     'database-name' = 'mydw', 
  16.     'table-name' = 'order_detail' 
  17. ); 

查询结果如下：

执行JOIN操作：

  1. SELECT 
  2.     od.id, 
  3.     oi.id order_id, 
  4.     oi.user_id, 
  5.     oi.province_id, 
  6.     od.sku_id, 
  7.     od.sku_name, 
  8.     od.sku_num, 
  9.     od.order_price, 
  10.     oi.create_time, 
  11.     oi.operate_time 
  12. FROM 
  13.    ( 
  14.     SELECT *  
  15.     FROM order_info 
  16.     WHERE  
  17.       order_status = '2'-- 已支付 
  18.    ) oi 
  19.    JOIN 
  20.   ( 
  21.     SELECT * 
  22.     FROM order_detail 
  23.   ) od  
  24.   ON oi.id = od.order_id; 

canal-json的操作实践

关于cannal的使用方式，可以参考我的另一篇文章：基于Canal与Flink实现数据实时增量同步(一)。我已经将下面的表通过canal同步到了kafka，具体格式为：

  1. { 
  2.     "data":[ 
  3.         { 
  4.             "id":"1", 
  5.             "region_name":"华北" 
  6.         }, 
  7.         { 
  8.             "id":"2", 
  9.             "region_name":"华东" 
  10.         }, 
  11.         { 
  12.             "id":"3", 
  13.             "region_name":"东北" 
  14.         }, 
  15.         { 
  16.             "id":"4", 
  17.             "region_name":"华中" 
  18.         }, 
  19.         { 
  20.             "id":"5", 
  21.             "region_name":"华南" 
  22.         }, 
  23.         { 
  24.             "id":"6", 
  25.             "region_name":"西南" 
  26.         }, 
  27.         { 
  28.             "id":"7", 
  29.             "region_name":"西北" 
  30.         } 
  31.     ], 
  32.     "database":"mydw", 
  33.     "es":1597128441000, 
  34.     "id":102, 
  35.     "isDdl":false, 
  36.     "mysqlType":{ 
  37.         "id":"varchar(20)", 
  38.         "region_name":"varchar(20)" 
  39.     }, 
  40.     "old":null, 
  41.     "pkNames":null, 
  42.     "sql":"", 
  43.     "sqlType":{ 
  44.         "id":12, 
  45.         "region_name":12 
  46.     }, 
  47.     "table":"base_region", 
  48.     "ts":1597128441424, 
  49.     "type":"INSERT" 
  50. } 

在SQL CLI中创建该canal-json格式的表：

  1. CREATE TABLE region ( 
  2.   id BIGINT, 
  3.   region_name STRING 
  4. ) WITH ( 
  5.  'connector' = 'kafka', 
  6.  'topic' = 'mydw.base_region', 
  7.  'properties.bootstrap.servers' = 'kms-3:9092', 
  8.  'properties.group.id' = 'testGroup', 
  9.  'format' = 'canal-json' , 
  10.  'scan.startup.mode' = 'earliest-offset'  
  11. ); 

查询结果如下：

changelog-json的操作实践

创建MySQL数据源

参见上面的order_info

Flink SQL Cli创建changelog-json表

  1. CREATE TABLE order_gmv2kafka ( 
  2.   day_str STRING, 
  3.   gmv DECIMAL(10, 5) 
  4. ) WITH ( 
  5.     'connector' = 'kafka', 
  6.     'topic' = 'order_gmv_kafka', 
  7.     'scan.startup.mode' = 'earliest-offset', 
  8.     'properties.bootstrap.servers' = 'kms-3:9092', 
  9.     'format' = 'changelog-json' 
  10. ); 
  11.  
  12. INSERT INTO order_gmv2kafka 
  13. SELECT DATE_FORMAT(create_time, 'yyyy-MM-dd') as day_str, SUM(total_amount) as gmv 
  14. FROM order_info 
  15. WHERE order_status = '2' -- 订单已支付 
  16. GROUP BY DATE_FORMAT(create_time, 'yyyy-MM-dd');  

查询表看一下结果：

再查一下kafka的数据：

  1. {"data":{"day_str":"2020-06-18","gmv":433},"op":"+I"} 

当将另外两个订单的状态order_status更新为2时，总金额=443+772+88=1293再观察数据：

再看kafka中的数据：

总结

本文主要介绍了基于FlinK构建实时数仓的技术点，并对其使用方式进行了详细描述，通过本文你或许对实时数仓和流批一体的应用会有一个深刻认识，希望本文对你有所帮助。

 

来源：大数据技术与数仓内容投诉

免责声明：

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题

  • 2024年上半年信息系统项目管理师第二批次真题及答案解析（完整版）

    难度     813人已做
    查看

  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看

  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看

  • 2024年上半年软考高项第一、二批次真题考点汇总（完整版）

    难度     435人已做
    查看

  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机

一文了解基于Flink1.12构建流批一体数仓的技术点

后端开发数仓Flink2024-12-03
位置：首页-资讯-后端开发
咦！没有更多了？去看看其它编程学习网 内容吧
首页课程
资料下载问答资讯