金融类网站设计欣赏/seo怎么学

背景

传统的入库任务一般借助于MapReduce或者Spark来写hive表，一般都是天级别最多小时级别的任务。随着实时性要求越来越高，传统的入库不太能满足需求。Flink完全基于流式处理，同时也支持了写Hive表。本文介绍一下如果通过FlinkSQL实现kafka数据入库hive，并能够实时可查。

Hive Catalog

由于写hive表必须基于hive catalog，所以需要注册hive catalog。同时可以在一个job内切换catalog，如果我们不想把kafka的source table注册到hive metastore里面，那么就可以使用memory catalog。

SQL

完整SQL如下

USE CATALOG hive_catalog;

USE mydb;--要写入的hive db

set table.sql-dialect=hive;

DROP TABLE IF EXISTS hive_table;

CREATE EXTERNAL TABLE hive_table (

data_timestamp BIGINT,

a STRING,

b STRING

) PARTITIONED BY (`day` STRING, `hour` STRING) STORED AS PARQUET

TBLPROPERTIES (

--这里支持filesystem connector的所有参数

'parquet.compression'='SNAPPY',--压缩算法

'sink.partition-commit.policy.kind' = 'metastore,success-file',--分区提交策略，自动创建分区和success文件

'sink.partition-commit.success-file.name' = '_SUCCESS'

);

USE CATALOG default_catalog; --可选，使用默认catalog，也可以使用hive catalog

set table.sql-dialect=default;--这里必须切换到default dialect

DROP TABLE IF EXISTS kafka_table;

CREATE TABLE kafka_table (

data_timestamp BIGINT,

a STRING,

b STRING

) WITH (

'connector' = 'kafka',

'topic' = 'topicx',

'properties.bootstrap.servers' = 'xxx:9092',

'properties.group.id' = 'topicx-groupid',

'scan.startup.mode' = 'latest-offset',

'format' = 'json'

);

insert into hive_catalog.mydb.hive_table --如果使用default catalog，这里必须使用全限定名，使用hive catalog则不需要

select

data_timestamp,

a,

b,

from_unixtime(data_timestamp/1000,'yyyy-MM-dd') as `day`,

from_unixtime(data_timestamp/1000,'HH') as `hour`

from kafka_table;

首先使用hive catalog+hive dialect创建hive表。

切换到default catalog+default dialect 创建kafka表。

insert into ... select ... from ...;

以上sql需要借助sql submit来提交。

补充

对于已有的hive表，同样也是可以写入的。但是得通过

alter table table_name set tblproperties('property_name'='new_value');语法将flink需要用到的属性设置进去。