Hive

概览

Hive 加载节点可以将数据写入 Hive。使用 Flink 方言,目前仅支持 Insert 操作,Upsert 模式下的数据会转换成 Insert 方式 目前暂时不支持使用 Hive 方言操作 Hive 表。

支持的版本

Load NodeVersion
HiveHive: 1.x, 2.x, 3.x

依赖

通过 Maven 引入 sort-connector-hive 构建自己的项目。 当然,你也可以直接使用 INLONG 提供的 jar 包。(sort-connector-hive)

Maven 依赖

  1. <dependency>
  2. <groupId>org.apache.inlong</groupId>
  3. <artifactId>sort-connector-hive</artifactId>
  4. <version>1.7.0</version>
  5. </dependency>

如何配置 Hive 数据加载节点

SQL API 的使用

使用 Flink SQL Cli :

  1. CREATE TABLE hiveTableName (
  2. id STRING,
  3. name STRING,
  4. uv BIGINT,
  5. pv BIGINT
  6. ) WITH (
  7. 'connector' = 'hive',
  8. 'default-database' = 'default',
  9. 'hive-version' = '3.1.2',
  10. 'hive-conf-dir' = 'hdfs://localhost:9000/user/hive/hive-site.xml'
  11. );

InLong Dashboard 方式

配置

在创建数据流时,选择数据落地为 ‘Hive’ 然后点击 ‘Add’ 来配置 Hive 的相关信息。

Hive Configuration

InLong Manager Client 方式

TODO: 未来版本支持

Hive 加载节点参数信息

参数是否必须默认值数据类型描述
connector
必须(none)String指定使用什么连接器,这里应该是 ‘hive’。
default-database
必须(none)String指定数据库名称。
hive-conf-dir
必须(none)String本地构建项目可以将hive-site.xml构建到 classpath 中,未来 Dashboard 将支持本地上传能力。 目前通用方式只支持配置已经上传文件后的 HDFS 路径。
sink.partition-commit.trigger
可选(none)String如果表是分区表,可以配置触发模式。如:(process-time)
partition.time-extractor.timestamp-pattern
可选(none)String如果表是分区表,可以配置时间戳。如:(yyyy-MM-dd)
sink.partition-commit.delay
可选(none)String如果表是分区表,可以配置延迟时间。如:(10s,20s,1m…)
sink.partition-commit.policy.kind
可选(none)String分区提交策略通知下游某个分区已经写完毕可以被读取了。 metastore:向 metadata 增加分区。仅 hive 支持 metastore 策略,文件系统通过目录结构管理分区; success-file:在目录中增加 ‘_success’ 文件; 上述两个策略可以同时指定:’metastore,success-file’。 custom:通过指定的类来创建提交策略, 支持同时指定多个提交策略:’metastore,success-file’。
inlong.metric.labels可选(none)Stringinlong metric 的标签值,该值的构成为groupId=[groupId]&streamId=[streamId]&nodeId=[nodeId]。

数据类型映射

Hive typeFlink SQL type
char(p)CHAR(p)
varchar(p)VARCHAR(p)
stringSTRING
booleanBOOLEAN
tinyintTINYINT
smallintSMALLINT
intINT
bigintBIGINT
floatFLOAT
doubleDOUBLE
decimal(p, s)DECIMAL(p, s)
dateDATE
timestamp(9)TIMESTAMP
bytesBINARY
arrayLIST
mapMAP
rowSTRUCT