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

化妆品的网站建设卖印花图案设计网站

化妆品的网站建设,卖印花图案设计网站,网站建设华网天下公司,贵阳能做网站的公司有哪些1.Flink数据源 Flink可以从各种数据源获取数据#xff0c;然后构建DataStream 进行处理转换。source就是整个数据处理程序的输入端。 数据集合数据文件Socket数据kafka数据自定义Source 2.案例 2.1.从集合中获取数据 创建 FlinkSource_List 类#xff0c;再创建个 Student 类…1.Flink数据源 Flink可以从各种数据源获取数据然后构建DataStream 进行处理转换。source就是整个数据处理程序的输入端。 数据集合数据文件Socket数据kafka数据自定义Source 2.案例 2.1.从集合中获取数据 创建 FlinkSource_List 类再创建个 Student 类姓名、年龄、性别三个属性就行反正测试用 package com.qiyu;import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.ArrayList;/*** author MR.Liu* version 1.0* data 2023-10-18 16:13*/ public class FlinkSource_List {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);ArrayListStudent clicks new ArrayList();clicks.add(new Student(Mary,25,1));clicks.add(new Student(Bob,26,2));DataStreamStudent stream env.fromCollection(clicks);stream.print();env.execute();} }运行结果 Student{nameMary, age25, sex1} Student{nameBob, age26, sex2} 2.2.从文件中读取数据 文件数据 spark hello world kafka spark hadoop spark package com.qiyu;import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** author MR.Liu* version 1.0* data 2023-10-18 16:31*/ public class FlinkSource_File {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamString stream env.readTextFile(input/words.txt);stream.print();env.execute();} }运行结果没做任何处理 spark hello world kafka spark hadoop spark 2.3.从Socket中读取数据 package com.qiyu;import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** author MR.Liu* version 1.0* data 2023-10-18 17:41*/ public class FlinkSource_Socket {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();// 2. 读取文本流DataStreamSourceString lineDSS env.socketTextStream(192.168.220.130,7777);lineDSS.print();env.execute();} }运行结果 服务器上执行 nc -lk 7777 疯狂输出 控制台打印结果  6 hello 7 world 2.4.从Kafka中读取数据 pom.xml 添加Kafka连接依赖 dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-kafka_${scala.binary.version}/artifactIdversion${flink.version}/version/dependency package com.qiyu;import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;import java.util.Properties;/*** author MR.Liu* version 1.0* data 2023-10-19 10:01*/ public class FlinkSource_Kafka {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);Properties properties new Properties();properties.setProperty(bootstrap.servers, hadoop102:9092);properties.setProperty(group.id, consumer-group);properties.setProperty(key.deserializer,org.apache.kafka.common.serialization.StringDeserializer);properties.setProperty(value.deserializer,org.apache.kafka.common.serialization.StringDeserializer);properties.setProperty(auto.offset.reset, latest);DataStreamSourceString stream env.addSource(new FlinkKafkaConsumerString(clicks, new SimpleStringSchema(), properties));stream.print(Kafka);env.execute();} }启动 zk 和kafka 创建topic bin/kafka-topics.sh --create --bootstrap-server hadoop102:9092 --replication-factor 1 --partitions 1 --topic clicks 生产者、消费者命令 bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic clicks bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic clicks --from-beginning 启动生产者命令后疯狂输入  运行java类运行结果和生产者输入的是一样的 Kafka flinks Kafka hadoop Kafka hello Kafka nihaop 2.5.从自定义Source中读取数据 大多数情况下前面几个数据源已经满足需求了。但是遇到特殊情况我们需要自定义的数据源。实现方式如下 1.编辑自定义源Source package com.qiyu;import org.apache.flink.streaming.api.functions.source.SourceFunction;import java.util.Calendar; import java.util.Random;/*** author MR.Liu* version 1.0* data 2023-10-19 10:37*//**** 主要实现2个方法 run() 和 cancel()*/ public class FlinkSource_Custom implements SourceFunctionStudent {// 声明一个布尔变量作为控制数据生成的标识位private Boolean running true;Overridepublic void run(SourceContextStudent sourceContext) throws Exception {Random random new Random(); // 在指定的数据集中随机选取数据String[] name {Mary, Alice, Bob, Cary};int[] sex {1,2};int age 0;while (running) {sourceContext.collect(new Student(name[random.nextInt(name.length)],sex[random.nextInt(sex.length)],random.nextInt(100)));// 隔 1 秒生成一个点击事件方便观测Thread.sleep(1000);}}Overridepublic void cancel() {running false;} }2.编写主程序 package com.qiyu;import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** author MR.Liu* version 1.0* data 2023-10-19 10:46*/ public class FlinkSource_Custom2 {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1); //有了自定义的 source function调用 addSource 方法DataStreamSourceStudent stream env.addSource(new FlinkSource_Custom());stream.print(SourceCustom);env.execute();} }运行主程序运行结果 SourceCustom Student{nameMary, age1, sex46} SourceCustom Student{nameCary, age2, sex52} SourceCustom Student{nameBob, age1, sex14} SourceCustom Student{nameAlice, age1, sex84} SourceCustom Student{nameAlice, age2, sex82} SourceCustom Student{nameCary, age1, sex28} .............
http://www.sadfv.cn/news/301274/

相关文章:

  • 网站建设劳务合同wordpress手机字体变大6
  • 图书馆登录系统网站建设代码网站建设需要经过哪几个步骤
  • wordpress 经典主题企业seo案例
  • 服务器安全防护最新外贸seo
  • 塑胶原料东莞网站建设技术支持网站编辑电子商务网站运营专员
  • 摄影网站设计说明书企业采购
  • 免费空间自助建站模板网站没有备案怎么做支付
  • 网站轮播图居中代码怎么写网站建设 客户需求
  • 百度做一个网站多少钱建网
  • 分享类网站怎么做网站运营和维护都是干什么的
  • 闵行区怎么样哈尔滨seo优化效果
  • 没有备案号的网站免费咨询养生顾问
  • 外贸品牌网站制作博创安泰网站建设
  • 做详情页到那个网站找模特素材阿里巴巴采购网官网
  • 备案期间关网站吗部队网站建设方案
  • 江苏分销网站建设怎么查询网站ftp地址
  • 西安在线网站网站建设回龙观
  • 如何建立一个网站要多少钱山东企业站点seo
  • 交互网站wordpress 图表
  • 专业网站开发费用自己做网站还是开淘宝
  • 怎么看一个网站用什么程序做的怎么做装球的网站
  • 赣州做网站的公司有哪家wordpress 纯静态
  • 网站布局wordpress 图文投票
  • 手机网站建设 移商动力iis配网站
  • 如何建立和设置公司网站做网站框架
  • 中国 网站服务器 租金买了网站模版怎么做
  • 番禺做网站系统wordpress 中文标签 404
  • 建站工具哪个好用python一句做网站
  • 济宁建设网站制作网站自己建机房
  • 自己做网站如何盈利服务器怎么做看视频的网站