手动安装 Hadoop 生态：Flink 配置与测试

本文是“基于已有镜像手动安装”路线的第二部分（Flink）。前置环境（启动容器、Hadoop 配置）见 01-手动安装：Spark 配置与测试。版本号沿用原始教程示例（Flink 1.15.1），如需与集成镜像一致请改用 README 中的版本（1.17.0）。

一、安装配置 Flink

1.1 安装 PyFlink

pip install apache-flink==1.15.1

如果不需要用到 HiveCatalog，可直接进入 二、测试 Flink 进行测试。

1.2 修改 JDK 版本

镜像里自带的是 JDK11，但 Hive3 是基于 Java8 编译的，基于 JDK11 运行时 PyFlink 连接 HiveCatalog 会报错。因此修改 JDK 版本为 8：

# 删除已有的 JDK11
rm -rf /usr/lib/jvm/

# 安装 JDK8
apt-get update
apt-get install -y openjdk-8-jdk

# 设置环境变量
vi ~/.bashrc

# java
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
export PATH=$PATH:$JAVA_HOME/bin

source ~/.bashrc

# 测试 java 命令
java -version

1.3 安装 Maven

PyFlink 连接 Hive Catalog 时要配置一些依赖 jar 包，可以通过 maven 一次性批量安装：

cd /opt/third
wget http://dlcdn.apache.org/maven/maven-3/3.8.6/binaries/apache-maven-3.8.6-bin.tar.gz

tar -xvzf apache-maven-3.8.6-bin.tar.gz
ln -s apache-maven-3.8.6 maven

# 配置 maven 镜像
vi maven/conf/settings.xml
# <mirrors> 内添加如下内容

    <!-- 华为云镜像 -->
    <mirror>
        <id>huaweimaven</id>
        <name>huawei maven</name>
        <url>https://mirrors.huaweicloud.com/repository/maven/</url>
        <mirrorOf>central</mirrorOf>
    </mirror>
    <!-- 阿里云镜像 -->
    <mirror>
        <id>nexus-aliyun</id>
        <mirrorOf>central</mirrorOf>
        <name>Nexus aliyun</name>
        <url>http://maven.aliyun.com/nexus/content/groups/public</url>
    </mirror>

# 设置环境变量
vi ~/.bashrc

# maven
export M2_HOME=/opt/third/maven
export PATH=$PATH:$M2_HOME/bin

source ~/.bashrc
mvn -v

1.4 批量安装依赖

mkdir -p /opt/third/flink

cd /opt/third/flink
mkdir lib
# 新建 flink-dep.xml 定义依赖的 jar（后续可直接扩展）
vi flink-dep.xml

<?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>com.flink.dep</groupId>
  <artifactId>flink-dep</artifactId>
  <version>1.0.0</version>

  <properties>
    <flink.version>1.15.1</flink.version>
    <hadoop.version>3.2.2</hadoop.version>
    <hive.version>3.1.2</hive.version>
    <scala.binary.version>2.12</scala.binary.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.hive</groupId>
      <artifactId>hive-exec</artifactId>
      <version>${hive.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-mapreduce-client-core</artifactId>
      <version>${hadoop.version}</version>
    </dependency>
  </dependencies>
</project>

# 运行命令下载依赖
mvn -f flink-dep.xml dependency:copy-dependencies -DoutputDirectory=/opt/third/flink/lib

集成镜像中也内置了这份 flink-dep.xml（见 images/jupyter-notebook/bigdata/conf/flink/flink-dep.xml），可直接参考。

1.5 上传 Hive 配置文件

在 JupyterLab 新建 Terminal 执行：

# 新建目录
mkdir -p /opt/third/hive/conf

在 Web 界面中上传 hive-site.xml 到 /opt/third/hive/conf 目录下。

二、测试 Flink

新建 Python3 Notebook 来测试。

2.1 本地运行

pyflink_sql.ipynb：

from pyflink.table import EnvironmentSettings, TableEnvironment

env_settings = EnvironmentSettings.in_streaming_mode()
t_env = TableEnvironment.create(env_settings)

t_env.execute_sql("""
    CREATE TABLE random_source(
        id BIGINT,
        data TINYINT
    ) WITH (
        'connector' = 'datagen',
        'fields.id.kind'='sequence',
        'fields.id.start'='1',
        'fields.id.end'='8',
        'fields.data.kind'='sequence',
        'fields.data.start'='4',
        'fields.data.end'='11'
    )
""")

t_env.execute_sql("""
    CREATE TABLE print_sink (
        id BIGINT,
        data_sum TINYINT
    ) WITH (
        'connector' = 'print'
    )
""")

t_env.execute_sql("""
    INSERT INTO print_sink
        SELECT id, sum(data) as data_sum FROM
            (SELECT id / 2 as id, data FROM random_source )
        WHERE id > 1
        GROUP BY id
""").wait()

2.2 本地运行（HiveCatalog）

元数据会存储到 Hive MetaStore 中，pyflink_sql_hivecatalog.ipynb：

import os
from pyflink.table import EnvironmentSettings, TableEnvironment

env_settings = EnvironmentSettings.in_streaming_mode()
t_env = TableEnvironment.create(env_settings)

flink_lib_path = "/opt/third/flink/lib"
jars = []
for file in os.listdir(flink_lib_path):
    if file.endswith('.jar'):
        jars.append(os.path.basename(file))
str_jars = ';'.join(['file://' + flink_lib_path + '/' + jar for jar in jars])
t_env.get_config().get_configuration().set_string("pipeline.jars", str_jars)

from pyflink.table.catalog import HiveCatalog

# Create a HiveCatalog
catalog_name = "hive"
default_database = "default"
catalog = HiveCatalog(catalog_name, default_database, "/opt/third/hive/conf")
t_env.register_catalog(catalog_name, catalog)
t_env.use_catalog(catalog_name)

t_env.execute_sql("DROP TABLE IF EXISTS random_source_pyflink")
t_env.execute_sql("""
    CREATE TABLE IF NOT EXISTS random_source_pyflink (
        id BIGINT,
        data TINYINT
    ) WITH (
        'connector' = 'datagen',
        'fields.id.kind'='sequence',
        'fields.id.start'='1',
        'fields.id.end'='8',
        'fields.data.kind'='sequence',
        'fields.data.start'='4',
        'fields.data.end'='11'
    )
""")

t_env.execute_sql("DROP TABLE IF EXISTS print_sink_pyflink")
t_env.execute_sql("""
    CREATE TABLE IF NOT EXISTS print_sink_pyflink  (
        id BIGINT,
        data_sum TINYINT
    ) WITH (
        'connector' = 'print'
    )
""")

t_env.execute_sql("""
    INSERT INTO print_sink_pyflink
        SELECT id, sum(data) as data_sum FROM
            (SELECT id / 2 as id, data FROM random_source_pyflink )
        WHERE id > 1
        GROUP BY id
""").wait()

上述操作完成后，可通过 pyspark 作业查看到新建的表 random_source_pyflink、print_sink_pyflink：

import os
import findspark

os.environ['SPARK_HOME'] = '/opt/third/spark'
findspark.init()

from pyspark.sql import SparkSession

if __name__ == "__main__":
    spark = SparkSession.builder \
        .appName('spark-hive-demo') \
        .config("hive.metastore.uris", "thrift://192.168.90.150:9083") \
        .enableHiveSupport() \
        .getOrCreate()

    spark.sql("show tables").show()