flink_sql_client 1.10 与 hive集成读取实时数据

很期待用纯sql的形式来处理流式数据，flink 1.10推出了生产可用的 Hive 集成，拥有了更强的流式 SQL 处理能力。这次我们就来尝试一下啦~~

【本文大纲】1、环境准备2、SQL Client与hive集成配置3、用SQL Client读取kafka数据

1、环境准备

相关软件版本：linux版本：centos 6.5Java版本：jdk1.8Hive版本：hive-2.3.4Hadoop版本：hadoop-2.7.3flink: flink-1.10.0scala:scala-2.11kafka:kafka_2.11-2.3.0

有关java、hive、hadoop的安装之前写过了： Hive源码系列（一）hive2.1.1 hadoop2.7.3环境搭建

下面准备一下flink,scala,kafka环境

1.1 scala安装

下载 scala-2.11.12.tgz

代码语言：javascript复制

tar -zxvf  scala-2.11.12.tgz ##解压scalaln -s flink-1.10.0 flink ##软链接vim /etc/profile ##设置环境变量

代码语言：javascript复制

代码语言：javascript复制

source /etc/profile ##生效

测试：

1.2 kafka安装

下载kafka_2.11-2.3.0.tgz

代码语言：javascript复制

tar -zxvf kafka_2.11-2.3.0.tgz  ##解压kafkaln -s kafka_2.11-2.3.0 kafka ##软链接
vim /etc/profile ##设置环境变量

代码语言：javascript复制

代码语言：javascript复制

source /etc/profile ##生效

启动kafka服务：

代码语言：javascript复制

zookeeper-server-start.sh $KAFKA_HOME/config/zookeeper.properties &kafka-server-start.sh $KAFKA_HOME/config/server.properties &

代码语言：javascript复制

创建测试的topic(flinktest):

代码语言：javascript复制

kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic flinktest
kafka-topics.sh --list --bootstrap-server localhost:9092 ##查看创建的topic

代码语言：javascript复制

分别启动生产者和 消费者测试一下：

代码语言：javascript复制

kafka-console-producer.sh --broker-list localhost:9092 --topic flinktest  ##生产者
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic flinktest --from-beginning ##消费者

如下图，说明生产数据没有问题：

以上，准备好了实时数据源，方便后面做测试用

1.3 flink安装

下载flink-1.10.0-bin-scala_2.11.tgz

代码语言：javascript复制

tar -zxvf  flink-1.10.0-bin-scala_2.11.tgz ##解压flinkln -s flink-1.10.0 flink ##软链接
vim /etc/profile ##设置环境变量

代码语言：javascript复制

代码语言：javascript复制

source /etc/profile ##生效

配置flink–Standalone模式：

代码语言：javascript复制

## flink-conf.yaml 文件配置vim $FLINK_HOME/conf/flink-conf.yaml  ##配置主节点的ip

代码语言：javascript复制

代码语言：javascript复制

## slavesvim $FLINK_HOME/conf/slaves ##配置从节点ip
##写入dataming

以上 flink单例模型配置完毕

2、SQL Client与hive集成配置

2.1 配制yaml文件

代码语言：javascript复制

cp $FLINK_HOME/conf/sql-client-defaults.yaml sql-client-hive.yamlvim $FLINK_HOME/conf/sql-client-hive.yaml

代码语言：javascript复制

2.2 加入依赖包

这一块是遇到问题最多的了

依赖hive相关包：

代码语言：javascript复制

$HIVE_HOME/lib/hive-exec-2.3.4.jar$HIVE_HOME/lib/hive-common-2.3.4.jar$HIVE_HOME/lib/hive-metastore-2.3.4.jar$HIVE_HOME/lib/hive-shims-common-2.3.4.jar$HIVE_HOME/lib/antlr-runtime-3.5.2.jar$HIVE_HOME/lib/datanucleus-api-jdo-4.2.4.jar$HIVE_HOME/lib/datanucleus-core-4.1.17.jar$HIVE_HOME/lib/datanucleus-rdbms-4.1.19.jar$HIVE_HOME/lib/javax.jdo-3.2.0-m3.jar$HIVE_HOME/lib/libfb303-0.9.3.jar$HIVE_HOME/lib/jackson-core-2.6.5.jar

代码语言：javascript复制

其它包：

代码语言：javascript复制

commons-cli-1.3.1.jarflink-connector-hive_2.11-1.10.0.jarflink-hadoop-compatibility_2.11-1.10.0.jarflink-shaded-hadoop2-uber-blink-3.2.4.jarflink-table-api-java-bridge_2.11-1.10.0.jarmysql-connector-java-5.1.9.jar

代码语言：javascript复制

将以上jar放入目录 $FLINK_HOME/lib

2.3 启动

代码语言：javascript复制

start-cluster.sh

3、用SQL Client读取kafka数据

3.1 启动sql client

代码语言：javascript复制

sql-client.sh embedded -d conf/sql-client-hive.yaml

3.2 创建表

代码语言：javascript复制

CREATE TABLE mykafka (name String, age Int) WITH (   'connector.type' = 'kafka',   'connector.version' = 'universal',   'connector.topic' = 'flinktest',   'connector.properties.zookeeper.connect' = 'localhost:2181',   'connector.properties.bootstrap.servers' = 'localhost:9092',   'format.type' = 'csv',   'update-mode' = 'append');

代码语言：javascript复制