flink1.13读kafka写入hive
作者：互联网
第一次写博客，flink自学了一下，网上查找案例，群里问大神，勉强把kafka写hive的流程走通了，在此记录一下，给后来者做个参考 ，其中有考虑不周或者错误的地方，请大家多多包涵和指正，嘻嘻。
代码
package cn.bywin
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, SqlDialect, Table, TableEnvironment, TableResult}
import org.apache.flink.table.catalog.hive.HiveCatalog
object HiveConnectDemo {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
val tableEnv = StreamTableEnvironment.create(env)
// 必须添加checkpoint
env.enableCheckpointing(30000,CheckpointingMode.EXACTLY_ONCE)
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
System.setProperty("HADOOP_USER_NAME","hdfs")
val name = "myhive"
val defaultDatabase = "stg"
// hive-site配置文件所在目录
val hiveConfDir = "/etc/hive/conf.cloudera.hive"
// val hiveConfDir = "/home/software/hive-2.1.1/conf"
// 注册hive的Catalog
val hive = new HiveCatalog(name, defaultDatabase, hiveConfDir)
tableEnv.registerCatalog("myhive", hive)
// 使用注册的catalog
tableEnv.useCatalog("myhive")
tableEnv.executeSql("drop table if exists hw_clzp2")
// 创建kafka连接
tableEnv.executeSql("""
|CREATE TABLE hw_clzp2 (
| alarmId string,
| alarmLevel string,
| `ts` TIMESTAMP(3) METADATA FROM 'timestamp',
| WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
|) WITH (
| 'connector' = 'kafka',
| 'topic' = 'zhsq_clzp',
| 'properties.bootstrap.servers' = '192.168.3.208:9092',
| 'properties.group.id' = 'testGroup',
| 'scan.startup.mode' = 'latest-offset',
| 'format' = 'json',
| 'json.fail-on-missing-field' = 'false',
| 'json.ignore-parse-errors' = 'true'
|)""".stripMargin)
// 切换到hive方言创建hive表,auto-compaction和compaction.file-size设置hive小文件合并
tableEnv.getConfig.setSqlDialect(SqlDialect.HIVE)
tableEnv.executeSql("drop table if exists zhsq_clzp2");
tableEnv.executeSql("""CREATE TABLE IF NOT EXISTS zhsq_clzp2
|(
|alarmId STRING,
|alarmLevel STRING
|)
|PARTITIONED BY (dt STRING, hr STRING)
|STORED AS parquet TBLPROPERTIES (
|'partition.time-extractor.timestamp-pattern'='$dt $hr:00:00',
|'sink.partition-commit.trigger'='partition-time',
|'sink.partition-commit.delay'='0s',
|'sink.partition-commit.policy.kind'='metastore,success-file',
|'auto-compaction'='true',
|'compaction.file-size'='128MB'
|)""".stripMargin);
// 切换到default方言
tableEnv.getConfig.setSqlDialect(SqlDialect.DEFAULT)
// 向Hive表中写入数据
tableEnv.executeSql("insert into zhsq_clzp2 select alarmId,alarmLevel,DATE_FORMAT(ts, 'yyyy-MM-dd'), DATE_FORMAT(ts, 'HH') from hw_clzp2")
}
}
说明
1.如果程序运行正常，最终会在hdfs相应目录看到有数据文件和success文件生成，如果没有success文件，请查看代码里checkpoint是否开启。
2.关于小文件合并问题，这里所说的合并并不是指一个分区内所有小文件都会合并成一个大文件，而是值一个checkpoint周期内的小文件会合并成一个文件，比如checkpoint周期是10分钟，那么每10分钟就会生成一个文件。
关于依赖
自己使用的依赖（里边有一些不需要的，我也懒得甄别了，大家自行去除）
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>cn.bywin</groupId>
<artifactId>flink1.13</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<flink.version>1.13.1</flink.version>
<scala.binary.version>2.11</scala.binary.version>
<scala.version>2.11.10</scala.version>
</properties>
<repositories>
<repository>
<id>scala-tools.org</id>
<name>Scala-Tools Maven2 Repository</name>
<url>http://scala-tools.org/repo-releases</url>
</repository>
<repository>
<id>cloudera</id>
<url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>scala-tools.org</id>
<name>Scala-Tools Maven2 Repository</name>
<url>http://scala-tools.org/repo-releases</url>
</pluginRepository>
</pluginRepositories>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<!-- <scope>provided</scope>-->
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<!-- flink-connector-jdbc -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<!-- <scope>provided</scope>-->
</dependency>
<!-- flink-connector-kafka -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<!-- mysql-->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.44</version>
</dependency>
<!-- hive-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>2.1.1</version>
<scope>provided</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/org.antlr/antlr-runtime -->
<dependency>
<groupId>org.antlr</groupId>
<artifactId>antlr-runtime</artifactId>
<version>3.5.2</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-json</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-csv</artifactId>
<version>${flink.version}</version>
</dependency>
</dependencies>
<build>
<sourceDirectory>src/main/scala</sourceDirectory>
<!-- <testSourceDirectory>src/test/scala</testSourceDirectory>-->
<plugins>
<plugin>
<groupId>org.scala-tools</groupId>
<artifactId>maven-scala-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.6.0</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.0.0</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<reporting>
<plugins>
<plugin>
<groupId>org.scala-tools</groupId>
<artifactId>maven-scala-plugin</artifactId>
<configuration>
<scalaVersion>${scala.version}</scalaVersion>
</configuration>
</plugin>
</plugins>
</reporting>
</project>
学习过程中主要参考了这篇文章：FLINK 1.12.2 读取KAFKA写入HIVE的完整示例
标签：scala,flink,hive,kafka,version,apache,org,flink1.13 来源： https://blog.csdn.net/qq_36062467/article/details/119203333