阿里云服务器25端口

1、 概述

1.1 案例介绍

Apache Flink 是一个开源的流处理框架，具有高吞吐、低延迟、可容错等特点，可同时支持批处理和流处理，为数据处理提供了强大而灵活的解决方案，Flink在 Docker中的应用场景主要是为了简化集群的部署和管理，特别是在开发、测试以及小规模生产环境中。使用 Docker 可以快速启动、停止和重启集群，避免手动配置和依赖管理的复杂性。

Flink实时统计功能可以应用在以下场景：

katago云服务器

实时数据清洗和转换：在数据进入存储或分析系统之前，需要对原始数据进行清洗和转换，以确保数据的质量和一致性。实时事件监测与告警：在实时监控系统中，当某些事件满足特定条件时触发告警。实时推荐系统；根据用户的实时行为和偏好，为用户提供个性化推荐。

本案例通过云主机进行Docker部署和安装Flink，在CodeArts IDE编辑器进行代码开发实现数据的实时统计。

通过实际操作，让大家深入了解如何方便快捷的使用Flink。在这个过程中，大家将学习到Docker的安装、Flink的安装部署以及简单的Flink代码开发，从而掌握 Flink 的基本使用方法，体验其在应用开发中的优势。

1.2 适用对象

企业个人开发者高校学生

1.3 案例时间

本案例总时长预计60分钟。

1.4 案例流程

说明：

登录云主机，安装Docker；在Docker安装Flink；安装JDK1.8；打开CodeArts IDE编写wordCount代码；代码打包到Flink运行。

1.5 资源总览

本案例花费总计0元。

最新案例动态，请查阅《基于华为开发者空间云开发环境部署Claude Code + KAT-Coder实现AI编程助手》「链接」。小伙伴快来领取华为开发者空间进行实操吧！

2、 Docker安装Flink实现数据实时统计

2.1 安装Docker

本案例中，使用Docker简化集群的部署和管理，提高开发效率、保证环境一致性、降低成本、提高安全性和可靠性，同时也支持复杂的架构和部署模式。

打开云主机命令行窗口输入以下命令，更新软件包。如果出现无法获得锁，锁正由unattended持有请等待一会重试。sudoapt updatesudoapt upgrade -y卸载旧版本Docker（如果已安装）。sudo apt-getremovedocker docker-engine docker.io containerd runc安装必要的依赖。sudoapt install apt-transport-https ca-certificates curl software-properties-common添加Docker的官方GPG密钥，如果添加失败可等待几分钟重试几次。sudocurl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo apt-key add -

注意：如果频繁失败，可以使用华为镜像站软件源替第4、5步骤中的Docker官方地址，命令如下：

sudo curl -fsSL https://mirrors.huaweicloud.com/docker-ce/linux/ubuntu/gpg | sudo apt-key add –sudoadd-apt-repository"deb [arch=amd64] https://mirrors.huaweicloud.com/docker-ce/linux/ubuntu $(lsb_release -cs) stable" 添加Docker的APT源。sudo add-apt-repository"deb [arch=amd64] https://download.docker.com/linux/ubuntu$(lsb_release -cs)stable"

执行命令后需要按ENTER键继续执行命令。

更新APT包索引。sudoapt update安装Docker CE。sudoapt install docker-ce验证Docker是否安装成功。sudo systemctlstatusdocker设置Docker自动启动。sudo systemctlenabledocker安装docker-compose。sudo apt-getinstall docker-compose

2.2 拉取Flink镜像

Apache Flink 是一个功能强大的流处理框架，适用于各种实时数据处理和分析场景，它提供了强大的功能和丰富的 API，支持分布式、高性能、低延迟和精确一次的处理，在现代数据处理领域发挥着重要的作用。

使用以下命令从 OBS下载指定版本的 Flink 镜像，并将镜像加载到本地的Docker镜像库中。sudowget https://dtse-mirrors.obs.cn-north-4.myhuaweicloud.com/case/0029/overseas/flink_image.tar sudo docker load -i flink_image.tar启动 Flink 集群

创建一个目录用于存放 Flink 集群的相关文件。

mkdir ~/flink &&cd~/flink创建 Docker Compose 文件

在~/flink目录下创建一个名为 docker-compose.yml 的文件。

vimdocker-compose.yml

文件内容如下：

version: 3services:jobmanager:image:flink:1.16.3-scala_2.12-java8hostname:jobmanagercontainer_name:jobmanagerports:-"8081:8081"environment:-|FLINK_PROPERTIES=jobmanager.rpc.address:jobmanagerparallelism.default:1command:jobmanagertaskmanager:image:flink:1.16.3-scala_2.12-java8hostname:taskmanagercontainer_name:taskmanagerenvironment:-|FLINK_PROPERTIES=jobmanager.rpc.address:jobmanagertaskmanager.numberOfTaskSlots:2parallelism.default:1depends_on:-jobmanagercommand:taskmanager 配置代理

在 Docker 的配置文件中添加华为镜像加速器。

sudovim /etc/docker/daemon.json

配置信息如下：

{"registry-mirrors": ["https://7046a839d8b94ca190169bc6f8b55644.mirror.swr.myhuaweicloud.com"] }

重启docker。

sudosystemctl restart docker启动 Flink 集群

通过以下命令启动 Flink 集群：

sudodocker-compose up -d

等待容器启动后，你可以通过访问 http://localhost:8081 来打开 Flink 的 Web 界面，以检查集群是否成功启动。

修改配置文件，保证日志正常打印

执行如下命令复制taskmanager下的docker-entrypoint.sh脚本。

sudodocker cp taskmanager:/docker-entrypoint.sh ./docker-entrypoint.sh

替换配置文件。

vimdocker-entrypoint.sh

配置文件如下：

!/usr/bin/env bashLicensed to the Apache Software Foundation (ASF) under oneor more contributor license agreements. See the NOTICE filedistributed with this work for additional informationregarding copyright ownership. The ASF licenses this fileto you under the Apache License, Version 2.0 (the"License"); you may not use this file except in compliancewith the License. You may obtain a copy of the License athttp://www.apache.org/licenses/LICENSE-2.0Unless required by applicable law or agreed to in writing, softwaredistributed under the License is distributed on an "AS IS" BASIS,WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.See the License for the specific language governing permissions andlimitations under the License.COMMAND_STANDALONE="standalone-job"COMMAND_HISTORY_SERVER="history-server"If unspecified, the hostname of the container is taken as the JobManager addressJOB_MANAGER_RPC_ADDRESS=${JOB_MANAGER_RPC_ADDRESS:-$(hostname -f)}CONF_FILE="${FLINK_HOME}/conf/flink-conf.yaml"drop_privs_cmd() {if[ $(id -u) != 0 ];thenDont need to drop privs if EUID != 0returnelif[ -x /sbin/su-exec ];thenAlpineechosu-exec flinkelseOthersechogosu flinkfi}copy_plugins_if_required() {if[ -z"$ENABLE_BUILT_IN_PLUGINS"];thenreturn0fiecho"Enabling required built-in plugins"fortarget_pluginin$(echo"$ENABLE_BUILT_IN_PLUGINS"| tr;);doecho"Linking${target_plugin}to plugin directory"plugin_name=${target_plugin%.jar}mkdir -p"${FLINK_HOME}/plugins/${plugin_name}"if[ ! -e"${FLINK_HOME}/opt/${target_plugin}"];thenecho"Plugin${target_plugin}does not exist. Exiting."exit1elseln -fs"${FLINK_HOME}/opt/${target_plugin}""${FLINK_HOME}/plugins/${plugin_name}"echo"Successfully enabled${target_plugin}"fidone}set_config_option() {localoption=$1localvalue=$2escape periods for usage in regular expressionslocalescaped_option=$(echo${option}| sed -e"s/\./\\\./g")either override an existing entry, or append a new oneifgrep -E"^${escaped_option}:.*""${CONF_FILE}"> /dev/null;thensed -i -e"s/${escaped_option}:.*/$option:$value/g""${CONF_FILE}"elseecho"${option}:${value}">>"${CONF_FILE}"fi}prepare_configuration() { set_config_option jobmanager.rpc.address${JOB_MANAGER_RPC_ADDRESS}set_config_option blob.server.port 6124 set_config_option query.server.port 6125if[ -n"${TASK_MANAGER_NUMBER_OF_TASK_SLOTS}"];thenset_config_option taskmanager.numberOfTaskSlots${TASK_MANAGER_NUMBER_OF_TASK_SLOTS}fiif[ -n"${FLINK_PROPERTIES}"];thenecho"${FLINK_PROPERTIES}">>"${CONF_FILE}"fienvsubst <"${CONF_FILE}">"${CONF_FILE}.tmp"&& mv"${CONF_FILE}.tmp""${CONF_FILE}"}maybe_enable_jemalloc() {if["${DISABLE_JEMALLOC:-false}"=="false"];thenJEMALLOC_PATH="/usr/lib/$(uname -m)-linux-gnu/libjemalloc.so"JEMALLOC_FALLBACK="/usr/lib/x86_64-linux-gnu/libjemalloc.so"if[ -f"$JEMALLOC_PATH"];thenexportLD_PRELOAD=$LD_PRELOAD:$JEMALLOC_PATHelif[ -f"$JEMALLOC_FALLBACK"];thenexportLD_PRELOAD=$LD_PRELOAD:$JEMALLOC_FALLBACKelseif["$JEMALLOC_PATH"="$JEMALLOC_FALLBACK"];thenMSG_PATH=$JEMALLOC_PATHelseMSG_PATH="$JEMALLOC_PATHand$JEMALLOC_FALLBACK"fiecho"WARNING: attempted to load jemalloc from$MSG_PATHbut the library couldnt be found. glibc will be used instead."fifi} maybe_enable_jemalloc copy_plugins_if_required prepare_configuration args=("$@")if["$1"="help"];thenprintf"Usage:$(basename "$0")(jobmanager|${COMMAND_STANDALONE}|taskmanager|${COMMAND_HISTORY_SERVER})\n"printf" Or$(basename "$0")help\n\n"printf"By default, Flink image adopts jemalloc as default memory allocator. This behavior can be disabled by setting the DISABLE_JEMALLOC environment variable to true.\n"exit0elif["$1"="jobmanager"];thenargs=("${args[@]:1}")echo"Starting Job Manager"exec$(drop_privs_cmd)"$FLINK_HOME/bin/jobmanager.sh"start-foreground"${args[@]}"elif["$1"=${COMMAND_STANDALONE}];thenargs=("${args[@]:1}")echo"Starting Job Manager"exec$(drop_privs_cmd)"$FLINK_HOME/bin/standalone-job.sh"start-foreground"${args[@]}"elif["$1"=${COMMAND_HISTORY_SERVER}];thenargs=("${args[@]:1}")echo"Starting History Server"exec$(drop_privs_cmd)"$FLINK_HOME/bin/historyserver.sh"start-foreground"${args[@]}"elif["$1"="taskmanager"];thenargs=("${args[@]:1}")echo"Starting Task Manager"$FLINK_HOME/bin/taskmanager.sh start"$@"fisleep 1exec/bin/bash -c"tail -f$FLINK_HOME/log/*.log"args=("${args[@]}")Running command in pass-through modeexec$(drop_privs_cmd)"${args[@]}"

修改后如下图：

将修改后的配置文件再上传到taskmanager。

sudodocker cp ./docker-entrypoint.sh taskmanager:/docker-entrypoint.sh

重启服务。

linux阿里云服务器管理

sudodocker-compose restart

查看服务状态。

sudodocker ps -a

2.3 安装Java环境

jdk1.8的安装包如下，请把压缩格式的文件jdk-8u391-linux-aarch64.tar.gz下载到云主机

复制下面链接到浏览器下载。

https://dtse-mirrors.obs.cn-north-4.myhuaweicloud.com/case/0001/jdk-8u391-linux-aarch64.tar.gz

把安装包上传到/home/developer/Downloads的目录下执行如下命令：

sudomkdir -p /usr/lib/jvm创建/usr/lib/jvm目录用来存放JDK文件sudo tar -zxvf /home/developer/Downloads/jdk-8u391-linux-aarch64.tar.gz -C /usr/lib/jvm把JDK文件解压到/usr/lib/jvm目录下

JDK文件解压缩以后，可以执行如下命令到/usr/lib/jvm目录查看一下：

cd/usr/lib/jvm ls

可以看到，在/usr/lib/jvm目录下有个jdk1.8.0_391目录。

下面继续执行如下命令，设置环境变量：

cd~ vim~/.bashrc

使用vim编辑器，打开了developer这个用户的环境变量配置文件，请在这个文件的开头位置，添加如下几行内容：

exportJAVA_HOME=/usr/lib/jvm/jdk1.8.0_391exportJRE_HOME=${JAVA_HOME}/jreexportCLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/libexportPATH=${JAVA_HOME}/bin:$PATH

保存.bashrc文件并退出vim编辑器。然后，继续执行如下命令让.bashrc文件的配置立即生效：

source~/.bashrc

这时，可以使用如下命令查看是否安装成功：

java-version

如果能够在屏幕上返回如下信息，则说明安装成功：

至此，就成功安装了Java环境。

2.4 代码开发

双击打开桌面上的CodeArts IDE for JAVA。

点击新建工程。

工程信息如下：

名称：自定义位置：默认构建系统：MavenJDK：1.8

配置settings.xml文件，在命令行执行：

vim/home/developer/.m2/settings.xml

将内容替换如下：

<settingsxmlns="http://maven.apache.org/SETTINGS/1.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/SETTINGS/1.0.0 http://maven.apache.org/xsd/settings-1.0.0.xsd"> <servers> <name>huaweiyun-maven <activeProfiles><activeProfile>dev