当前位置： 首页 > news >正文

做网站卖什么东西好企业营销网站建设系统

news 2025/7/9 16:50:52
做网站卖什么东西好,企业营销网站建设系统,公司注册资金变更怎么办理,网站开发的标准目录 前提条件 基本准备 批处理API实现WordCount 流处理API实现WordCount 数据源是文件 数据源是socket文本流 打包 提交到集群运行 命令行提交作业 Web UI提交作业 上传代码到gitee 前提条件 Windows安装好jdk8、Maven3、IDEA Linux安装好Flink集群，可…

目录

前提条件

基本准备

批处理API实现WordCount

流处理API实现WordCount

数据源是文件

数据源是socket文本流

打包

提交到集群运行

命令行提交作业

Web UI提交作业

上传代码到gitee

前提条件

Windows安装好jdk8、Maven3、IDEA

Linux安装好Flink集群，可参考：CentOS7安装flink1.17完全分布式
 

基本准备

创建项目

使用IDEA创建一个新的Maven项目，项目名称，例如：flinkdemo

添加依赖

在项目的pom.xml文件中添加Flink的依赖。

	<properties><flink.version>1.17.1</flink.version></properties><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>${flink.version}</version></dependency></dependencies>

刷新依赖

刷新依赖后，能看到相关依赖如下

刷新依赖过程需要等待一些时间来下载相关依赖。

如果依赖下载慢，可以设置阿里云仓库镜像：

 1.设置maven的settings.xml

在</mirrors>上面一行添加阿里云仓库镜像

	<mirror><id>alimaven</id><name>aliyun maven</name><url>http://maven.aliyun.com/nexus/content/groups/public/</url><mirrorOf>central</mirrorOf>        </mirror>

2.IDEA设置maven

数据准备

在工程的根目录下，新建一个data文件夹

并在data文件夹下创建文本文件words.txt

内容如下

hello world
hello java
hello flink

新建包

右键src/main下的java，新建Package

填写包名org.example，包名与groupId的内容一致。

批处理API实现WordCount

在org.exmaple下新建wc包及BatchWordCount类

填写wc.BatchWordCount

效果如下

BatchWordCount.java代码如下：

package org.example.wc;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;public class BatchWordCount {public static void main(String[] args) throws Exception {// 1. 创建执行环境ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();// 2. 从文件读取数据 按行读取DataSource<String> lineDS = env.readTextFile("data/words.txt");// 3. 转换数据格式FlatMapOperator<String, Tuple2<String, Long>> wordAndOne = lineDS.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {@Overridepublic void flatMap(String line, Collector<Tuple2<String, Long>> out) throws Exception {String[] words = line.split(" ");for (String word : words) {out.collect(Tuple2.of(word,1L));}}});// 4. 按照 word 进行分组UnsortedGrouping<Tuple2<String, Long>> wordAndOneUG = wordAndOne.groupBy(0);// 5. 分组内聚合统计AggregateOperator<Tuple2<String, Long>> sum = wordAndOneUG.sum(1);// 6. 打印结果sum.print();}
}

运行程序，查看结果

注意，以上代码的实现方式是基于DataSet API的，是批处理API。而Flink本身是流批统一的处理架构，批量的数据集本质上也是流，没有必要用两套不同的API来实现。从Flink 1.12开始，官方推荐直接使用DataStream API，在提交任务时通过将执行模式设为BATCH来进行批处理：

$ flink run -Dexecution.runtime-mode=BATCH BatchWordCount.jar

流处理API实现WordCount

数据源是文件

在org.example.wc包下新建Java类StreamWordCount，代码如下：

package org.example.wc;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;public class StreamWordCount {public static void main(String[] args) throws Exception {// 1. 创建流式执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 2. 读取文件DataStreamSource<String> lineStream = env.readTextFile("input/words.txt");// 3. 转换、分组、求和，得到统计结果SingleOutputStreamOperator<Tuple2<String, Long>> sum = lineStream.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {@Overridepublic void flatMap(String line, Collector<Tuple2<String, Long>> out) throws Exception {String[] words = line.split(" ");for (String word : words) {out.collect(Tuple2.of(word, 1L));}}}).keyBy(data -> data.f0).sum(1);// 4. 打印sum.print();// 5. 执行env.execute();}
}

运行结果

与批处理程序BatchWordCount的区别：

  • 创建执行环境的不同，流处理程序使用的是StreamExecutionEnvironment。

  • 转换处理之后，得到的数据对象类型不同。

  • 分组操作调用的是keyBy方法，可以传入一个匿名函数作为键选择器（KeySelector），指定当前分组的key是什么。

  • 代码末尾需要调用env的execute方法，开始执行任务。

数据源是socket文本流

流处理的输入数据通常是流数据，将StreamWordCount代码中读取文件数据的readTextFile方法，替换成读取socket文本流的方法socketTextStream。

在org.example.wc包下新建Java类SocketStreamWordCount，代码如下：

package org.example.wc;import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;public class SocketStreamWordCount {public static void main(String[] args) throws Exception {// 1. 创建流式执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 2. 读取文本流：node2表示发送端主机名（根据实际情况修改）、7777表示端口号DataStreamSource<String> lineStream = env.socketTextStream("node2", 7777);// 3. 转换、分组、求和，得到统计结果SingleOutputStreamOperator<Tuple2<String, Long>> sum = lineStream.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {String[] words = line.split(" ");for (String word : words) {out.collect(Tuple2.of(word, 1L));}}).returns(Types.TUPLE(Types.STRING, Types.LONG)).keyBy(data -> data.f0).sum(1);// 4. 打印sum.print();// 5. 执行env.execute();}
}

进入node2终端，如果没有nc命令，需要先安装nc命令，安装nc命令如下：

[hadoop@node2 ~]$ sudo yum install nc -y

开启nc监听

[hadoop@node2 ~]$ nc -lk 7777

IDEA中，运行SocketStreamWordCount程序。

往7777端口发送数据，例如发送hello world

控制台输出

继续往7777端口发送数据，例如发送hello flink

控制台输出

停止SocketStreamWordCount程序。

按Ctrl+c停止nc命令。

打包

这里的打包是将写好的程序打成jar包。

点击IDEA右侧的Maven，按住Ctrl键同时选中clean和package（第一次打包可以只选中package），点击执行打包。

打包成功后，看到如下输出信息，生成的jar包在项目的target目录下

提交到集群运行

把jar包提交到flink集群运行有两种方式：

1.通过命令行提交作业   

2.通过Web UI提交作业

命令行提交作业

将jar包上传Linux

启动flink集群
[hadoop@node2 ~]$ start-cluster.sh 
Starting cluster.
Starting standalonesession daemon on host node2.
Starting taskexecutor daemon on host node2.
Starting taskexecutor daemon on host node3.
Starting taskexecutor daemon on host node4.
​
开启nc监听
[hadoop@node2 ~]$ nc -lk 7777
​
命令提交作业

开启另一个node2终端，使用flink run命令提交作业到flink集群

[hadoop@node2 ~]$ flink run -m node2:8081 -c org.example.wc.SocketStreamWordCount flinkdemo-1.0-SNAPSHOT.jar

-m指定提交到的JobManager，-c指定程序入口类。

发送测试数据

在nc监听终端，往7777端口发送数据

查看结果
Web UI查看结果

浏览器访问

node2:8081

看到正在运行的作业如下

查看结果

继续发送测试数据

在nc终端继续发送数据

Web UI刷新结果

命令行查看结果

打开新的node2终端，查看结果

[hadoop@node2 ~]$ cd $FLINK_HOME/log
[hadoop@node2 log]$ ls
flink-hadoop-client-node2.log                 flink-hadoop-standalonesession-0-node2.out
flink-hadoop-standalonesession-0-node2.log    flink-hadoop-taskexecutor-0-node2.log
flink-hadoop-standalonesession-0-node2.log.1  flink-hadoop-taskexecutor-0-node2.log.1
flink-hadoop-standalonesession-0-node2.log.2  flink-hadoop-taskexecutor-0-node2.log.2
flink-hadoop-standalonesession-0-node2.log.3  flink-hadoop-taskexecutor-0-node2.log.3
flink-hadoop-standalonesession-0-node2.log.4  flink-hadoop-taskexecutor-0-node2.log.4
flink-hadoop-standalonesession-0-node2.log.5  flink-hadoop-taskexecutor-0-node2.out
[hadoop@node2 log]$ cat flink-hadoop-taskexecutor-0-node2.out 
(hello,1)
(flink,1)
(hello,2)
(world,1)
​

取消flink作业

点击Cancel Job取消作业 

停止nc监听

按Ctrl+c停止nc命令

Web UI提交作业

开启nc监听

开启nc监听发送数据

[hadoop@node2 ~]$ nc -lk 7777

Web UI提交作业

浏览器访问

node2:8081

点击Submit New Job

点击Add New

选择flink作业jar包所在路径

点击jar包名称

填写相关内容，点击Submit提交作业

Entry Class填写运行的主类，例如：org.example.wc.SocketStreamWordCount

Parallesim填写作业的并行度，例如：1

提交后，在Running Jobs里看到运行的作业

发送测试数据

往7777端口发送数据

查看结果

继续发送测试数据

刷新结果

取消作业

停止nc监听

按住Ctrl+c停止nc命令

关闭flink集群
[hadoop@node2 ~]$ stop-cluster.sh 
Stopping taskexecutor daemon (pid: 2283) on host node2.
Stopping taskexecutor daemon (pid: 1827) on host node3.
Stopping taskexecutor daemon (pid: 1829) on host node4.
Stopping standalonesession daemon (pid: 1929) on host node2.

上传代码到gitee

登录gitee

https://gitee.com/

注意：如果还没有gitee账号，需要先注册；如果之前没有设置过SSH公钥，需要先设置SSH公钥。

创建仓库

提交代码

使用IDEA提交代码

提示有警告，忽略警告，继续提交

提交成功后，IDEA显示如下

刷新浏览器查看gitee界面，看到代码已上传成功

完成！enjoy it!

http://www.hrbkazy.com/news/19187.html

相关文章：

  • wdcp搭建网站百度推广获客方法
  • 做个平台网站怎么做北京百度网站排名优化
  • wordpress制作图片站站长工具seo综合查询下载
  • 怎么在网页中加入图片黑帽seo教程
  • 网站宣传的方式会计培训班
  • ps软件下载电脑版要钱吗北京seo公司有哪些
  • 装修公司做自己网站西安百度推广优化
  • 烟台提供网站设计制作泰安网站推广优化
  • 运营型网站衡阳seo快速排名
  • 小说网站怎么做流量吗seo优化的方法有哪些
  • 响应式网站设计制作免费做网站怎么做网站
  • 做店铺装修的公司网站google引擎免费入口
  • 网站开发作业代做网络安全
  • 门户网站建设jz190网络营销的类型
  • 江苏外协机械加工网适合seo软件
  • 东营做网站优化价格排名点击软件怎样
  • 西部数码做的网站打不开网站策划是做什么的
  • 网站怎么做搜索栏网络营销策划书的结构
  • 申请网站空间怎么做网站运营策划书
  • 雅虎做网站推广浙江百度推广开户
  • 网站建设台州网站外链分析工具
  • 学院 网站 两学一做北京网站优化步骤
  • 国外公共空间设计网站百度学术论文查重免费
  • 网络推广网站电话seo搜索引擎优化步骤
  • 可以看男男做的视频网站google chrome谷歌浏览器
  • 现在的网站推广是怎么做的网站seo优化是什么意思
  • wordpress html音乐全网seo优化电话
  • 北京便宜做网站互动营销名词解释
  • 布吉做棋牌网站建设哪家便宜seo优化在线诊断
  • 国内设计的企业网站百度一下点击搜索