美文网首页Flink实战
Flink实战之Kafka To Hive

Flink实战之Kafka To Hive

作者: 〇白衣卿相〇 | 来源:发表于2020-12-10 20:17 被阅读0次

    背景

    传统的入库任务一般借助于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;
    
    1. 首先使用hive catalog+hive dialect创建hive表。
    2. 切换到default catalog+default dialect 创建kafka表。
    3. insert into ... select ... from ...;

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

    补充

    对于已有的hive表,同样也是可以写入的。但是得通过
    alter table table_name set tblproperties('property_name'='new_value');语法将flink需要用到的属性设置进去。

    相关文章

      网友评论

        本文标题:Flink实战之Kafka To Hive

        本文链接:https://www.haomeiwen.com/subject/mptewktx.html