Flink1.11.2 使用SQL方式消费kafka写入Hive测试用例实践
前言提示：这里可以添加本文要记录的大概内容：例如：随着人工智能的不断发展，机器学习这门技术也越来越重要，很多人都开启了学习机器学习，本文就介绍了机器学习的基础内容。提示：以下是本篇文章正文内容，下面案例可供参考一、pandas是什么？示例：pandas 是基于NumPy 的一种工具，该工具是为了解决数据分析任务而创建的。二、使用步骤1.引入库代码如下（示例）：import numpy as np
·
前言
本文基于Flink1.11.2 的稳定版本探讨flink实时写入Hive的技术可行性，下面是个本地测试的案例可供参考。
一、Flink ETL SQL化思路
我们有很多实时数据是存储在kafka中，如何按照分区低延迟的高效存储在Hive数仓中以便于近实时分析是我们现在一个普遍诉求。
这里暂不涉及修改的记录，使用场景局限在某些日志类型，如涉及更新修改的应考察数据湖方案。Flink在1.11版本中已经实现了流批统一，
Table API & SQL 统一了 DataStream API 和 DataSet API ，如没有特殊业务场景，个人觉得可以多使用 Table API &SQL . 我们知道
TableEnvironment 是 flink 中集成 Table API 和 SQL 的核心概念，所有对表的操作都基于 TableEnvironment
- 注册catalog
- 在内部 catalog 中注册表
- 执行 SQL 查询
- 注册用户自定义函数
- 将 DataStream 或 DataSet 转换为表等
二、初始化执行环境
1.引入相关依赖（maven）
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.zm</groupId>
<artifactId>FlinkETL</artifactId>
<version>1.0-SNAPSHOT</version>
<packaging>jar</packaging>
<name>Flink Quickstart Job</name>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<flink.version>1.11.2</flink.version>
<hive.version>2.1.1</hive.version>
<java.version>1.8</java.version>
<scala.binary.version>2.11</scala.binary.version>
<maven.compiler.source>${java.version}</maven.compiler.source>
<maven.compiler.target>${java.version}</maven.compiler.target>
</properties>
<dependencies>
<!-- Apache Flink dependencies -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-json</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>${flink.version}</version>
</dependency>
<!--使用Java编程语言支持DataStream / DataSet API的Table＆SQL API-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<!--必须将以下依赖项添加到项目中才能使用Table API和SQL来定义管道-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<!--通过自定义函数，格式等扩展表生态系统的通用模块-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-common</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<!-- Flink kafka connector: kafka版本大于1.0.0可以直接使用通用的连接器 -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.11</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-shaded-hadoop-2-uber</artifactId>
<version>2.7.5-10.0</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-hive_2.11</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_2.11</artifactId>
<version>1.11.2</version>
<scope>provided</scope>
</dependency>
<!-- Hive Dependency -->
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>${hive.version}</version>
<scope>provided</scope>
</dependency>
<!-- Add logging framework, to produce console output when running in the IDE. -->
<!-- These dependencies are excluded from the application JAR by default. -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.7</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
<scope>provided</scope>
</dependency>
</dependencies>
<build>
<plugins>
<!-- Java Compiler -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.1</version>
<configuration>
<source>${java.version}</source>
<target>${java.version}</target>
</configuration>
</plugin>
<!-- We use the maven-shade plugin to create a fat jar that contains all necessary dependencies. -->
<!-- Change the value of <mainClass>...</mainClass> if your program entry point changes. -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.0.0</version>
<executions>
<!-- Run shade goal on package phase -->
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<artifactSet>
<excludes>
<exclude>org.apache.flink:force-shading</exclude>
<exclude>com.google.code.findbugs:jsr305</exclude>
<exclude>org.slf4j:*</exclude>
<exclude>log4j:*</exclude>
</excludes>
</artifactSet>
<filters>
<filter>
<!-- Do not copy the signatures in the META-INF folder.
Otherwise, this might cause SecurityExceptions when using the JAR. -->
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
<transformers>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.zm.StreamingJob</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<!-- This profile helps to make things run out of the box in IntelliJ -->
<!-- Its adds Flink's core classes to the runtime class path. -->
<!-- Otherwise they are missing in IntelliJ, because the dependency is 'provided' -->
<profiles>
<profile>
<id>add-dependencies-for-IDEA</id>
<activation>
<property>
<name>idea.version</name>
</property>
</activation>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>compile</scope>
</dependency>
</dependencies>
</profile>
</profiles>
</project>
2.准备资源文件
【kerberos认证相关】
krb5.conf
xxx.keytab
【Hive访问相关资源文件】
core-site.xml
hdfs-site.xml
hive-site.xml
mapred-site.xml
yarn-site.xml
三、示例代码
1、安全认证
// kerberos
String confPath = "/conf/uat/krb5.conf";
System.setProperty("java.security.krb5.conf", confPath);
String keyPath = "/conf/uat/xx.keytab";
UserGroupInformation.loginUserFromKeytab("xx@FAYSON.COM", keyPath);2、初始化执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.setParallelism(1);
EnvironmentSettings envSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env, envSettings);
tableEnvironment.getConfig().getConfiguration().set(ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(10));
tableEnvironment.getConfig().getConfiguration().set(ExecutionCheckpointingOptions.CHECKPOINTING_MODE, CheckpointingMode.EXACTLY_ONCE);3、使用HiveCatalog构造TableEnvironment
String DEFAULT_CATALOG = "default_catalog";
String HIVE_CATALOG = "myhive";
String DEFAULT_DATABASE = "tmp";
String HIVE_CONF_DIR = "/path/resources";
Catalog catalog = new HiveCatalog(HIVE_CATALOG, DEFAULT_DATABASE, HIVE_CONF_DIR);
tableEnvironment.registerCatalog(HIVE_CATALOG, catalog);
tableEnvironment.useCatalog("myhive");4、执行具体sql
// 构造kafka流表
TableResult tableResult = tableEnvironment.executeSql("DROP TABLE IF EXISTS ods_k_table_earliest");
tableResult.print();
TableResult tableResult2 = tableEnvironment.executeSql(
"CREATE TABLE ods_k_table_earliest (\n" +
" user_id STRING,\n" +
" order_amount DOUBLE,\n" +
" log_ts TIMESTAMP(3),\n" +
" WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND\n" +
" ) WITH (\n" +
" 'connector.type' = 'kafka',\n" +
" 'connector.version' = 'universal',\n " +
" 'connector.topic' = 't_kafka_02',\n" +
" 'connector.properties.bootstrap.servers' = 'xxx:9092',\n" +
" 'connector.properties.zookeeper.connect' = 'xxx:2181',\n" +
" 'connector.properties.group.id' = 'group_test_01',\n" +
" 'connector.startup-mode' = 'earliest-offset',\n" +
" 'format.type' = 'json'\n" +
" )");
tableResult2.print();
// 构造Hive目标表
TableResult t1 = tableEnvironment.executeSql("DROP TABLE IF EXISTS t_kafka_03");
t1.print();
TableResult tableResult3 = tableEnvironment.executeSql(
"CREATE TABLE t_kafka_03 (\n" +
" user_id STRING,\n" +
" order_amount DOUBLE,\n" +
" log_ts TIMESTAMP(3),\n" +
" WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND\n" +
") WITH (\n" +
" 'connector'='kafka',\n" +
" 'topic'='t_kafka_03',\n" +
" 'scan.startup.mode'='latest-offset',\n" +
" 'properties.bootstrap.servers'='xxx:9092',\n" +
" 'properties.group.id' = 'testGroup_01',\n" +
" 'format'='json'\n" +
")");
tableResult3.print();
// 数据写入，无需提交，使用executeSql就能触发
tableEnvironment.getConfig().setSqlDialect(SqlDialect.HIVE);
TableResult tableResult4 = tableEnvironment.executeSql(
"INSERT INTO ods_hive_table " +
" SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH') FROM t_kafka_03");
5、Kafka客户端数据模拟写入
{"user_id":"a1111","order_amount":11.0,"log_ts":"2020-06-29 12:12:12"}
{"user_id":"a1111","order_amount":11.0,"log_ts":"2020-06-29 12:15:00"}
{"user_id":"a1111","order_amount":11.0,"log_ts":"2020-06-29 12:20:00"}
{"user_id":"a1111","order_amount":11.0,"log_ts":"2020-06-29 12:30:00"}
{"user_id":"a1111","order_amount":13.0,"log_ts":"2020-06-29 12:32:00"}
{"user_id":"a1112","order_amount":15.0,"log_ts":"2020-11-26 12:12:12"}
{"user_id":"a1112","order_amount":15.0,"log_ts":"2020-11-26 12:15:00"}
{"user_id":"a1112","order_amount":15.0,"log_ts":"2020-11-26 12:20:00"}
{"user_id":"a1112","order_amount":15.0,"log_ts":"2020-11-26 12:30:00"}
{"user_id":"a1112","order_amount":15.0,"log_ts":"2020-11-26 12:32:00"}6、Hive输出
hive> select * from ods_hive_table;
OK
a1111 11.0 2020-06-29 12
a1111 11.0 2020-06-29 12
a1111 11.0 2020-06-29 12
a1111 11.0 2020-06-29 12
a1111 13.0 2020-06-29 12
a1112 15.0 2020-11-26 12
a1112 15.0 2020-11-26 12
a1112 15.0 2020-11-26 12
a1112 15.0 2020-11-26 12
a1112 15.0 2020-11-26 12
Time taken: 0.514 seconds, Fetched: 10 row(s)
hive> show partitions ods_hive_table;
OK
dt=2020-06-29/hr=12
dt=2020-11-26/hr=12
Time taken: 0.091 seconds, Fetched: 4 row(s)
总结
本次的测试数据旨在模拟FlinkSQL通过消费kafka写入Hive的完整流程，还未得到生产环境的验证，我这边测试的结果是消费延迟很低，或许和我测试规模有关，
我这里不清楚当大批量数据过来是不会有很明显的延迟，但是生产中如果能够接受一定的时间延迟，比如半小时，1个小时，作为近实时的数据分析是有FlinkSql还是可以支持的。
后面主要考虑模拟的场景更丰富些，添加一些监控措施，让实时的流程序更健壮。
Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。
更多推荐
0
0- 0
已为社区贡献4条内容
所有评论(0)