大白话理解:每个流式的Dstream,其实底层也是通过rdd来操作,transform就是将Dstream转成rdd,就可以和其他的普通的rdd做各种运算操作
1 新建zidian.txt文件
24##少林寺欲将功夫融入足球培训运动员
25##北京警方捣毁侵害公民信息团伙抓获299人
26##北京发布雷电黄色预警傍晚至夜间将有大到暴雨
27##聚划算成清仓专用问题产品充斥
28##七夕前夜英仙座流星雨助兴我国处于最佳观测区
30##报告称中国取代英国成世界第5大常规武器出口国
2 在158.158.4.49上,运行nc -lk 9998
输入数据格式为24 good
3 主程序代码
package cn.taobao;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.StorageLevels;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;
import java.util.regex.Pattern;
public class Transform_Test
{
public static void main(String[] args) throws Exception {
// StreamingContext 编程入口
JavaStreamingContext ssc = new JavaStreamingContext(
/*
按照Cpu最多Cores来设置线程数
*/
"local[*]",
"TwoSocketJoin_Test",
Durations.seconds(4),
System.getenv("SPARK_HOME"),
JavaStreamingContext.jarOfClass(Transform_Test.class.getClass()));
ssc.sparkContext().setLogLevel("ERROR");
//从文件中返回 JavaPairRDD数据
String file="C:\\Users\\need\\Desktop\\zidian.txt";
JavaRDD<String> stringJavaRDD = ssc.sparkContext().textFile(file);
JavaPairRDD<String, String> keyvalueRDD = stringJavaRDD.mapToPair(new PairFunction<String, String, String>() {
@Override
public Tuple2<String, String> call(String s) throws Exception {
String[] arr = s.split("##");
return new Tuple2<>(arr[0], arr[1]);
}
});
//数据接收器(Receiver)
//创建一个接收器(JavaReceiverInputDStream),这个接收器接收一台机器上的某个端口通过socket发送过来的数据并处理
JavaReceiverInputDStream<String> socketTextStream_1 = ssc.socketTextStream(
"158.158.4.49", 9998, StorageLevels.MEMORY_AND_DISK_SER);
JavaPairDStream<String, String> javaPairDStream_1 = socketTextStream_1.mapToPair(new PairFunction<String, String, String>() {
@Override
public Tuple2<String, String> call(String s) throws Exception {
String[] s1 = Pattern.compile(" ").split(s);
return new Tuple2<>(s1[0], s1[1]);
}
});
//流式Dstream和rdd做运算
JavaPairDStream<String, Tuple2<String, String>> stringTuple2JavaPairDStream = javaPairDStream_1.transformToPair(aa -> aa.join(keyvalueRDD));
stringTuple2JavaPairDStream.print();
//显式的启动数据接收
ssc.start();
try {
//来等待计算完成
ssc.awaitTermination();
} catch (Exception e) {
e.printStackTrace();
} finally {
ssc.close();
}
}
}
运行结果如下
-------------------------------------------
Time: 1619588148000 ms
-------------------------------------------
(33,(keyi,汽车高速爆胎致车祸鉴定机构回避轮胎质量问题))
-------------------------------------------
Time: 1619588152000 ms
-------------------------------------------