Flink-读Kafka写Hive表

介绍

在实时大数据处理中，Apache Flink是一个广泛使用的流处理引擎，而Apache Kafka则是一个高吞吐量的分布式消息队列。在许多场景下，我们需要将Kafka中的数据实时写入Hive表中，以便进行后续处理和分析。本篇博客将介绍如何使用Flink来读取Kafka数据并将其写入Hive表。

准备工作

在开始之前，确保已经安装好了Flink、Kafka和Hive。同时，也需要创建好要使用的Kafka topic和Hive表。

步骤

步骤一：引入依赖

首先，在你的Flink项目中，需要添加相应的依赖。在pom.xml文件中添加以下依赖：

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.11</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.11</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-hive_2.11</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

确保${flink.version}已经正确配置为你使用的Flink版本号。

步骤二：编写Flink程序

接下来，我们需要编写一个Flink程序来读取Kafka数据并写入Hive表。下面是一个简单的示例代码：

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.types.Row;

import java.util.Properties;

public class KafkaToHiveJob {

    public static void main(String[] args) throws Exception {

        // 设置Flink的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        // 设置Kafka相关属性
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "flink-group");

        // 创建Kafka消费者
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties);

        // 从Kafka读取数据
        DataStream<String> kafkaStream = env.addSource(kafkaConsumer);

        // 将数据转换为Table
        tEnv.connect(
                new Kafka()
                        .version("0.11")
                        .topic("topic")
                        .property("bootstrap.servers", "localhost:9092")
                        .property("group.id", "flink-group")
                        .startFromLatest()
        )
                .withFormat(new Json().failOnMissingField(false))
                .inAppendMode()
                .registerTableSource("kafkaTable");

        Table kafkaTable = tEnv.from("kafkaTable");

        // 将Table写入Hive表
        String catalogName = "hive_catalog";
        String databaseName = "your_database";
        String hiveTableName = "your_table";
        String hiveTablePath = "/user/hive/" + databaseName + "/" + hiveTableName;
        String hiveTableNameWithDB = databaseName + "." + hiveTableName;

        Catalog hiveCatalog = new HiveCatalog(catalogName, databaseName, hiveTableNamePath);
        tEnv.registerCatalog(catalogName, hiveCatalog);

        tEnv.insertInto(hiveTableNameWithDB, kafkaTable);

        // 执行Flink程序
        env.execute("KafkaToHiveJob");
    }
}

步骤三：运行程序

将上述代码保存为一个Java类，并通过Flink的命令行工具或者IDE来运行该类。

$ bin/flink run -c com.example.KafkaToHiveJob my-flink-project-1.0.jar

步骤四：验证结果

在程序顺利执行后，你应该能够在Hive中看到数据已经被成功写入了指定的表中。你可以通过Hive的查询语句来验证数据是否被正常写入。

结论

通过上述步骤，我们成功地使用了Flink来读取Kafka数据并将其写入Hive表中。通过灵活配置Flink和相关依赖，我们可以轻松实现从实时数据源到数据湖的数据传输。

本文来自极简博客，作者：夜色温柔，转载请注明原文链接：Flink-读Kafka写Hive表