flinksource同步_Hive终于等来了Flink

简介&＃xff1a; Flink 社区在集成 Hive 功能方面付出很多&＃xff0c;目前进展也比较顺利&＃xff0c;最近 Flink 1.10.0 RC1 版本已经发布&＃xff0c;感兴趣的读者可以进行调研和验证功能。

作者&＃xff1a;Jason
Apache Spark 什么时候开始支持集成 Hive 功能&＃xff1f;笔者相信只要使用过 Spark 的读者&＃xff0c;应该都会说这是很久以前的事情了。
那 Apache Flink 什么时候支持与 Hive 的集成呢&＃xff1f;读者可能有些疑惑&＃xff0c;还没有支持吧&＃xff0c;没用过&＃xff1f;或者说最近版本才支持&＃xff0c;但是功能还比较弱。
其实比较也没啥意义&＃xff0c;不同社区发展的目标总是会有差异&＃xff0c;而且 Flink 在真正的实时流计算方面投入的精力很多。不过笔者想表达的是&＃xff0c;Apache Hive 已经成为数据仓库生态系统的焦点&＃xff0c;它不仅是一个用于大数据分析和 ETL 的 SQL 引擎&＃xff0c;也是一个数据管理平台&＃xff0c;所以无论是 Spark&＃xff0c;还是 Flink&＃xff0c;或是 Impala、Presto 等&＃xff0c;都会积极地支持集成 Hive 的功能。
的确&＃xff0c;对真正需要使用 Flink 访问 Hive 进行数据读写的读者会发现&＃xff0c;Apache Flink 1.9.0 版本才开始提供与 Hive 集成的功能。不过&＃xff0c;值得欣慰的是&＃xff0c;Flink 社区在集成 Hive 功能方面付出很多&＃xff0c;目前进展也比较顺利&＃xff0c;最近 Flink 1.10.0 RC1 版本已经发布&＃xff0c;感兴趣的读者可以进行调研和验证功能。
架构设计
首先&＃xff0c;笔者基于社区公开的资料以及博客&＃xff0c;概括性地讲解 Flink 集成 Hive 的架构设计。
Apache Flink 与 Hive 集成的目的&＃xff0c;主要包含了元数据和实际表数据的访问。
元数据
为了访问外部系统的元数据&＃xff0c;Flink 刚开始提供了 ExternalCatalog 的概念。但是 ExternalCatalog 的定义非常不完整&＃xff0c;基本处于不可用的状态。Flink 1.10 版本正式删除了 ExternalCatalog API (FLINK-13697)&＃xff0c;这包括:

• ExternalCatalog&＃xff08;以及所有依赖的类&＃xff0c;比如 ExternalTable&＃xff09;
• SchematicDescriptor、MetadataDescriptor 和 StatisticsDescriptor

针对 ExternalCatalog 的问题&＃xff0c;Flink 社区提出了一套全新的 Catalog 接口&＃xff08;new Catalog API&＃xff09;来取代现有的 ExternalCatalog。新的 Catalog 实现的功能包括&＃xff1a;

• 能够支持数据库、表、分区等多种元数据对象
• 允许在一个用户 Session 中维护多个 Catalog 实例&＃xff0c;从而支持同时访问多个外部系统
• Catalog 以可插拔的方式接入 Flink&＃xff0c;允许用户提供自定义的实现

下图展示了新的 Catalog API 的总体架构&＃xff1a;

创建 TableEnvironment 的时候会同时创建一个 CatalogManager&＃xff0c;负责管理不同的 Catalog 实例。TableEnvironment 通过 Catalog 来为 Table API 和 SQL Client 用户提供元数据服务。
val settings &＃61; EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build() val tableEnv &＃61; TableEnvironment.create(settings) val name &＃61; "myhive" val defaultDatabase &＃61; "mydatabase" val hiveConfDir &＃61; "/opt/hive-conf"// a local path val version &＃61; "2.3.4" val hive &＃61; newHiveCatalog(name, defaultDatabase, hiveConfDir, version) tableEnv.registerCatalog("myhive", hive) // set the HiveCatalog as the current catalog of the session tableEnv.useCatalog("myhive")
目前 Catalog 有两个实现&＃xff0c;GenericInMemoryCatalog 和 HiveCatalog。其中 GenericInMemoryCatalog 保持了原有的 Flink 元数据管理机制&＃xff0c;将所有元数据保存在内存中。而 HiveCatalog 会与一个 Hive Metastore 的实例连接&＃xff0c;提供元数据持久化的能力。要使用 Flink 与 Hive 进行交互&＃xff0c;用户需要配置一个 HiveCatalog&＃xff0c;并通过 HiveCatalog 访问 Hive 中的元数据。
另一方面&＃xff0c;HiveCatalog 也可以用来处理 Flink 自身的元数据&＃xff0c;在这种场景下&＃xff0c;HiveCatalog 仅将 Hive Metastore 作为持久化存储使用&＃xff0c;写入 Hive Metastore 中的元数据并不一定是 Hive 所支持的格式。一个 HiveCatalog 实例可以同时支持这两种模式&＃xff0c;用户无需为管理 Hive 和 Flink 的元数据创建不同的实例。
另外&＃xff0c;通过设计 HiveShim 来支持不同版本的 Hive Metastore&＃xff0c;具体支持的 Hive 版本列表&＃xff0c;请参考官方文档。
表数据
Flink 提供了 Hive Data Connector 来读写 Hive 的表数据。Hive Data Connector 尽可能的复用了 Hive 本身的 Input/Output Format 和 SerDe 等类&＃xff0c;这样做的好处一方面是减少了代码重复&＃xff0c;更重要的是可以最大程度的保持与 Hive 的兼容&＃xff0c;即 Flink 写入的数据 Hive 可以正常读取&＃xff0c;并且反之亦然。
集成 Hive 功能
Flink 与 Hive 集成的功能在 1.9.0 版本中作为试用功能发布&＃xff0c;存在不少使用的局限性&＃xff0c;但是不久将发布的 Flink 1.10 稳定版本会更加完善集成 Hive 的功能并应用到企业场景中。
为了让读者提前体验 Flink 1.10 集成 Hive 的功能&＃xff0c;笔者会基于 Cloudera CDH 编译 Flink 1.10.0 RC1 版本并进行较为完整的测试。
环境信息
CDH 版本&＃xff1a;cdh5.16.2
Flink 版本&＃xff1a;release-1.10.0-rc1
Flink 使用了 RC 版本&＃xff0c;仅供测试&＃xff0c;不建议用于生产环境。
目前 Cloudera Data Platform 正式集成了 Flink 作为其流计算产品&＃xff0c;非常方便用户使用。
CDH 环境开启了 Sentry 和 Kerberos。
下载并编译 Flink
$ wget https://http://github.com/apache/flink/archive/release-1.10.0-rc1.tar.gz $ tar zxvf release-1.10.0-rc1.tar.gz $ cd flink-release-1.10.0-rc1/ $ mvn clean install -DskipTests-Pvendor-repos -Dhadoop.version&＃61;2.6.0-cdh5.16.2
不出意外的话&＃xff0c;编译到 flink-hadoop-fs 模块时&＃xff0c;会报如下错误&＃xff1a;
[ERROR] Failed to execute goal on project flink-hadoop-fs: Could not resolve dependencies for project org.apache.flink:flink-hadoop-fs:jar:1.10.0: Failed to collect dependencies at org.apache.flink:flink-shaded-hadoop-2:jar:2.6.0-cdh5.16.2-9.0: Failed to read artifact descriptor for org.apache.flink:flink-shaded-hadoop-2:jar:2.6.0-cdh5.16.2-9.0: Could not transfer artifact org.apache.flink:flink-shaded-hadoop-2:pom:2.6.0-cdh5.16.2-9.0 from/to HDPReleases (https://repo.hortonworks.com/content/repositories/releases/): Remote host closed connection during handshake: SSL peer shut down incorrectly
编译中遇到 flink-shaded-hadoop-2 找不到的问题&＃xff0c;其实查看 Maven 仓库会发现&＃xff0c;根本原因是 CDH 的 flink-shaded-hadoop-2 的 jar 包在 Maven 中央仓库是没有对应的编译版本&＃xff0c;所以需要先对 Flink 依赖的 flink-shaded-hadoop-2 进行打包&＃xff0c;再进行编译。■ 解决 flink-shaded-hadoop-2 问题

• 获取 flink-shaded 源码

git clone https://github.com/apache/flink-shaded.git

• 切换依赖的版本分支

根据上面报错时提示缺少的版本切换对应的代码分支&＃xff0c;即缺少的是 9.0 版本的 flink-shaded-hadoop-2&＃xff1a;
git checkout release-9.0

• 配置 CDH Repo 仓库

修改 flink-shaded 项目中的 pom.xml&＃xff0c;添加 CDH maven 仓库&＃xff0c;否则编译时找不到 CDH 相关的包。
在 ... 中添加如下内容&＃xff1a;
vendor-repos vendor-repos cloudera-releases https://repository.cloudera.com/artifactory/cloudera-repos true false

• 编译 flink-shaded

开始执行编译&＃xff1a;
mvn clean install -DskipTests-Drat.skip&＃61;true-Pvendor-repos -Dhadoop.version&＃61;2.6.0-cdh5.16.2
建议通过科学上网方式编译&＃xff0c;如果读者遇到一些网络连接的问题&＃xff0c;可以试着重试或者更换依赖组件的仓库地址。
编译成功后&＃xff0c;就会把 flink-shaded-hadoop-2-uber-2.6.0-cdh5.16.2-9.0.jar 安装在本地 maven 仓库&＃xff0c;如下为编译的最后日志&＃xff1a;
Installing /Users/.../source/flink-shaded/flink-shaded-hadoop-2-uber/target/flink-shaded-hadoop-2-uber-2.6.0-cdh5.16.2-9.0.jar to /Users/.../.m2/repository/org/apache/flink/flink-shaded-hadoop-2-uber/2.6.0-cdh5.16.2-9.0/flink-shaded-hadoop-2-uber-2.6.0-cdh5.16.2-9.0.jar
Installing /Users/.../source/flink-shaded/flink-shaded-hadoop-2-uber/target/dependency-reduced-pom.xml to /Users/.../.m2/repository/org/apache/flink/flink-shaded-hadoop-2-uber/2.6.0-cdh5.16.2-9.0/flink-shaded-hadoop-2-uber-2.6.0-cdh5.16.2-9.0.pom
重新编译 Flink
mvn clean install -DskipTests-Pvendor-repos -Dhadoop.version&＃61;2.6.0-cdh5.16.2
漫长的等待过程&＃xff0c;读者可以并行做其他事情。
编译过程中&＃xff0c;如果不出意外的话&＃xff0c;会看到类似下面的错误信息&＃xff1a;
[INFO] Running &＃39;npm ci --cache-max&＃61;0 --no-save&＃39; in /Users/xxx/Downloads/Flink/flink-release-1.10.0-rc1/flink-release-1.10.0-rc1/flink-runtime-web/web-dashboard [WARNING] npm WARN prepare removing existing node_modules/ before installation [ERROR] WARN registry Unexpected warning for https://registry.npmjs.org/: Miscellaneous Warning ECONNRESET: request to https://registry.npmjs.org/mime/-/mime-2.4.0.tgz failed, reason: read ECONNRESET [ERROR] WARN registry Using stale package data from https://registry.npmjs.org/ due to a request error during revalidation. [ERROR] WARN registry Unexpected warning for https://registry.npmjs.org/: Miscellaneous Warning ECONNRESET: request to https://registry.npmjs.org/optimist/-/optimist-0.6.1.tgz failed, reason: read ECONNRESET
可以看到&＃xff0c; flink-runtime-web 模块引入了对 frontend-maven-plugin 的依赖&＃xff0c;需要安装 node、npm 和依赖组件。
如果没有通过科学上网&＃xff0c;可以修改 flink-runtime-web/pom.xml 文件&＃xff0c;添加 nodeDownloadRoot 和 npmDownloadRoot 的信息&＃xff1a;
com.github.eirslett frontend-maven-plugin 1.6 install node and npm install-node-and-npm https://registry.npm.taobao.org/dist/ https://registry.npmjs.org/npm/-/ v10.9.0 npm install npm ci --cache-max&＃61;0 --no-save true npm run build npm run build web-dashboard
编译成功后&＃xff0c;Flink 安装文件位于 flink-release-1.10.0-rc1/flink-dist/target/flink-1.10.0-bin 目录下&＃xff0c;打包并上传到部署到节点&＃xff1a;
$ cd flink-dist/target/flink-1.10.0-bin $ tar zcvf flink-1.10.0.tar.gz flink-1.10.0
部署和配置
Flink 部署比较简单&＃xff0c;解压缩包即可。另外可以设置软链接、环境变量等&＃xff0c;笔者不再介绍。
Flink 的核心配置文件是 flink-conf.yaml&＃xff0c;一个典型的配置如下&＃xff1a;
jobmanager.rpc.address: localhost jobmanager.rpc.port: 6123 jobmanager.heap.size: 2048m taskmanager.heap.size: 1024m taskmanager.numberOfTaskSlots: 4 parallelism.default: 1 high-availability: zookeeper high-availability.storageDir:hdfs:///user/flink110/recovery high-availability.zookeeper.quorum: zk1:2181,zk2:2181,zk3:2181 state.backend: filesystem state.checkpoints.dir: hdfs:///user/flink110/checkpoints state.savepoints.dir:hdfs:///user/flink110/savepoints jobmanager.execution.failover-strategy: region rest.port: 8081 taskmanager.memory.preallocate: false classloader.resolve-order: parent-first security.kerberos.login.use-ticket-cache: true security.kerberos.login.keytab:/home/flink_user/flink_user.keytab security.kerberos.login.principal: flink_user jobmanager.archive.fs.dir:hdfs:///user/flink110/completed-jobs historyserver.web.address: 0.0.0.0 historyserver.web.port: 8082 historyserver.archive.fs.dir:hdfs:///user/flink110/completed-jobs historyserver.archive.fs.refresh-interval: 10000
笔者只罗列了一些常见的配置参数&＃xff0c;读者根据实际情况修改。配置参数其实还是比较容易理解的&＃xff0c;以后结合实战的文章再进行详细讲解。■ 集成 Hive 配置的依赖
如果要使用 Flink 与 Hive 集成的功能&＃xff0c;除了上面的配置外&＃xff0c;用户还需要添加相应的依赖&＃xff1a;

• 如果需要使用 SQL Client&＃xff0c;则需要将依赖的 jar 拷贝到 Flink 的 lib 目录中
• 如果需要使用 Table API&＃xff0c;则需要将相应的依赖添加到项目中&＃xff08;如 pom.xml&＃xff09;

org.apache.flink flink-connector-hive_2.11 1.11-SNAPSHOT provided org.apache.flink flink-table-api-java-bridge_2.11 1.11-SNAPSHOT provided org.apache.hive hive-exec ${hive.version} provided
笔者主要介绍使用 SQL Client 的方式&＃xff0c;由于使用的 CDH 版本为 5.16.2&＃xff0c;其中 Hadoop 版本为 2.6.0&＃xff0c;Hive 版本为 1.1.0&＃xff0c;所以需要将如下 jar 包拷贝到 flink 部署家目录中的 lib 目录下&＃xff1a;

• Flink 的 Hive connector

flink-connector-hive2.11-1.10.0.jar
flink-hadoop-compatibility2.11-1.10.0.jar
flink-orc_2.11-1.10.0.jar
flink-release-1.10.0-rc1/flink-connectors/flink-hadoop-compatibility/target/flink-hadoop-compatibility_2.11-1.10.0.jar flink-release-1.10.0-rc1/flink-connectors/flink-connector-hive/target/flink-connector-hive_2.11-1.10.0.jar flink-release-1.10.0-rc1/flink-formats/flink-orc/target/flink-orc_2.11-1.10.0.jar

• Hadoop 依赖

flink-shaded-hadoop-2-uber-2.6.0-cdh5.16.2-9.0.jar
flink-shaded/flink-shaded-hadoop-2-uber/target/flink-shaded-hadoop-2-uber-2.6.0-cdh5.16.2-9.0.jar

• Hive 依赖

hive-exec-1.1.0-cdh5.16.2.jar
hive-metastore-1.1.0-cdh5.16.2.jar
libfb303-0.9.3.jar
/opt/cloudera/parcels/CDH/lib/hive/lib/hive-exec-1.1.0-cdh5.16.2.jar /opt/cloudera/parcels/CDH/lib/hive/lib/hive-metastore-1.1.0-cdh5.16.2.jar /opt/cloudera/parcels/CDH/lib/hive/lib/libfb303- 0.9.3.jar
其中 flink-shaded-hadoop-2-uber 包含了 Hive 对于 Hadoop 的依赖。如果不用 Flink 提供的包&＃xff0c;用户也可以将集群中使用的 Hadoop 包添加进来&＃xff0c;不过需要保证添加的 Hadoop 版本与 Hive 所依赖的版本是兼容的。
依赖的 Hive 包&＃xff08;即 hive-exec 和 hive-metastore&＃xff09;也可以使用用户集群中 Hive 所提供的 jar 包&＃xff0c;详情请见支持不同的 Hive 版本。
Flink 部署的节点要添加 Hadoop、Yarn 以及 Hive 的客户端。■ 配置 HiveCatalog
多年来&＃xff0c;Hive Metastore 在 Hadoop 生态系统中已发展成为事实上的元数据中心。许多公司在其生产中有一个单独的 Hive Metastore 服务实例&＃xff0c;以管理其所有元数据&＃xff08;Hive 元数据或非 Hive 元数据&＃xff09;。
如果同时部署了 Hive 和 Flink&＃xff0c;那么通过 HiveCatalog 能够使用 Hive Metastore 来管理 Flink 的元数据。
如果仅部署 Flink&＃xff0c;HiveCatalog 就是 Flink 开箱即用提供的唯一持久化的 Catalog。如果没有持久化的 Catalog&＃xff0c;那么使用 Flink SQL CREATE DDL 时必须在每个会话中重复创建像 Kafka 表这样的元对象&＃xff0c;这会浪费大量时间。HiveCatalog 通过授权用户只需要创建一次表和其他元对象&＃xff0c;并在以后的跨会话中非常方便地进行引用和管理。
如果要使用 SQL Client 时&＃xff0c;用户需要在 sql-client-defaults.yaml 中指定自己所需的 Catalog&＃xff0c;在 sql-client-defaults.yaml 的 catalogs 列表中可以指定一个或多个 Catalog 实例。
以下的示例展示了如何指定一个 HiveCatalog&＃xff1a;
execution: planner: blink type: streaming ... current-catalog: myhive # set the HiveCatalog as the current catalog of the session current-database: mydatabase catalogs: - name: myhive type: hive hive-conf-dir: /opt/hive-conf # contains hive-site.xml hive-version:2.3.4
其中&＃xff1a;

• name 是用户给每个 Catalog 实例指定的名字&＃xff0c;Catalog 名字和 DB 名字构成了 FlinkSQL 中元数据的命名空间&＃xff0c;因此需要保证每个 Catalog 的名字是唯一的。
• type 表示 Catalog 的类型&＃xff0c;对于 HiveCatalog 而言&＃xff0c;type 应该指定为 hive。
• hive-conf-dir 用于读取 Hive 的配置文件&＃xff0c;用户可以将其设定为集群中 Hive 的配置文件目录。
• hive-version 用于指定所使用的 Hive 版本。

指定了 HiveCatalog 以后&＃xff0c;用户就可以启动 sql-client&＃xff0c;并通过以下命令验证 HiveCatalog 已经正确加载。
Flink SQL> show catalogs; default_catalog myhive Flink SQL> use catalog myhive;
其中 show catalogs 会列出加载的所有 Catalog 实例。需要注意的是&＃xff0c;除了用户在 sql-client-defaults.yaml 文件中配置的 Catalog 以外&＃xff0c;FlinkSQL 还会自动加载一个 GenericInMemoryCatalog 实例作为内置的 Catalog&＃xff0c;该内置 Catalog 默认名字为 default_catalog。
读写 Hive 表
设置好 HiveCatalog 以后就可以通过 SQL Client 或者 Table API 来读写 Hive 中的表了。
假设 Hive 中已经有一张名为 mytable 的表&＃xff0c;我们可以用以下的 SQL 语句来读写这张表。■ 读数据
Flink SQL> show catalogs; myhive default_catalog Flink SQL> use catalog myhive; Flink SQL> show databases; default Flink SQL> show tables; mytable Flink SQL> describe mytable; root |-- name: name |-- type: STRING |-- name: value |-- type: DOUBLE Flink SQL> SELECT * FROM mytable; name value __________ __________ Tom 4.72 John 8.0 Tom 24.2 Bob. 3.14 Bob 4.72 Tom 34.9 Mary 4.79 Tiff 2.72 Bill 4.33 Mary 77.7■ 写数据
Flink SQL> INSERT INTO mytable SELECT &＃39;Tom&＃39;, 25; Flink SQL> INSERT OVERWRITE mytable SELECT &＃39;Tom&＃39;, 25; # 静态分区 Flink SQL> INSERT OVERWRITE myparttable PARTITION (my_type&＃61;&＃39;type_1&＃39;, my_date&＃61;&＃39;2019-08-08&＃39;) SELECT &＃39;Tom&＃39;, 25; # 动态分区 Flink SQL> INSERT OVERWRITE myparttable SELECT &＃39;Tom&＃39;, 25, &＃39;type_1&＃39;, &＃39;2019-08-08&＃39;; # 静态分区和动态分区 Flink SQL> INSERT OVERWRITE myparttable PARTITION (my_type&＃61;&＃39;type_1&＃39;) SELECT &＃39;Tom&＃39;, 25, &＃39;2019-08-08&＃39;;
总结
在本文中&＃xff0c;笔者首先介绍了 Flink 与 Hive 集成功能的架构设计&＃xff0c;然后从源码开始编译&＃xff0c;解决遇到的一些问题&＃xff0c;接着部署和配置 Flink 环境以及集成 Hive 的具体操作过程&＃xff0c;最后参考官方的案例&＃xff0c;对 Hive 表进行读写操作。
后续&＃xff0c;笔者会结合生产环境的实际使用情况&＃xff0c;讲解通过 Flink SQL 来操作 Hive。参考&＃xff1a;

• https://ci.apache.org/projects/flink/flink-docs-release-1.10/flinkDev/building.html
• https://ververica.cn/developers/flink1-9-hive/

SQL 分布式计算 Java Hadoop API Apache Maven HIVE 流计算 Spark