spack stream简介简单来说Spark Streaming 是 Apache Spark 框架中用于处理“实时数据流”的组件。你可以把它理解成一个“流水线处理器”能让数据像流水一样一边到达、一边就被处理而不是像传统批处理那样攒成一大堆再统一处理。它的核心价值是将实时数据流无界数据转换为一系列小的“微批次”有界数据进行处理。传统批处理好比快递站攒够一车货再发车时效性差但效率高。Spark Streaming好比快递站每隔1分钟就发一趟车每辆车处理这1分钟内到的货。这既保证了近实时的低延迟通常是秒级又复用了Spark强大的批处理引擎。常见的应用场景包括实时网站流量统计、金融交易风控、物联网设备数据监控、实时日志分析等。离线计算和流式计算介绍计算模式核心特点代表框架与技术离线计算 (Batch/Offline)处理静态、有界、海量的数据集注重高吞吐量和准确性对响应时间要求不高通常为分钟级、小时级甚至天级。Hadoop MapReduce、Hive、Spark Core流式计算 (Streaming/Real-time)处理动态、无界、持续产生的数据流注重低延迟秒级、毫秒级和实时响应能立即对数据进行分析和处理。Apache Flink、Apache Storm、Spark StreamingSpark Streaming-Java版本代码导入依赖dependencies dependency groupIdorg.apache.spark/groupId artifactIdspark-streaming_2.12/artifactId version3.3.1/version /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.3.1/version /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-streaming-kafka-0-10_2.12/artifactId version3.3.1/version /dependency /dependenciesSparkStreaming 基础环境模板/** * SparkStreaming Java 版本基础环境演示代码 * Spark对流式计算的核心运行环境、调度逻辑做了完整封装 */ public class SparkStreaming01_Env { public static void main(String[] args) throws InterruptedException { // 1. 创建Spark核心配置对象用于设置应用运行参数 SparkConf conf new SparkConf(); // 设置运行模式为本地多线程模式*代表自动匹配本机CPU核心数仅本地调试使用 conf.setMaster(local[*]); // 设置当前Spark应用名称会展示在SparkUI、YARN任务列表中 conf.setAppName(sparkStreaming); /** * 2. 创建SparkStreaming上下文对象流式程序核心入口 * 参数1Spark配置conf * 参数2Duration(1000)批次间隔1000毫秒1秒每1秒生成一批数据执行计算 * 该对象封装了流式任务调度、数据采集、批处理、资源管理全部核心能力 */ JavaStreamingContext jsc new JavaStreamingContext(conf, new Duration(1000)); // 缺失核心逻辑区域 // TODO 这里必须编写1.数据源读取 2.数据转换计算 3.输出行动算子(print/foreachRDD等) // 如果缺少数据源和输出算子调用start()时Spark会校验失败抛出 No output operations 异常 // // 3. 启动流式任务调度器开始持续采集、处理数据流 // 该方法仅启动后台调度线程不会阻塞主线程执行完会立刻向下运行 jsc.start(); // 4. 阻塞主线程让流式程序长期持续运行不会执行完main方法就退出进程 // 只有手动关闭程序、程序异常崩溃、主动调用stop()才会解除阻塞 jsc.awaitTermination(); } }网络数据流socket处理演示下面我们展示了监听本机的9999端口获取数据public static void main(String[] args) throws InterruptedException { SparkConf conf new SparkConf(); conf.setMaster(local[*]); conf.setAppName(sparkStreaming); JavaStreamingContext jsc new JavaStreamingContext(conf, new Duration(5000)); //todo: 通过环境对象获取Socket数据源获取数据模型进行数据处理 JavaReceiverInputDStreamString socketDS jsc.socketTextStream(localhost, 9999); //将获取的数据直接进行打印 socketDS.print(); jsc.start(); //阻塞代码他会一直阻塞在这里 jsc.awaitTermination(); }使用工具演示网络数据流的传输。使用命令进行启动指定端口号nc -lp 9999kafka数据源处理public static void main(String[] args) throws InterruptedException { SparkConf conf new SparkConf(); conf.setMaster(local[*]); conf.setAppName(sparkStreaming); JavaStreamingContext jsc new JavaStreamingContext(conf, new Duration(5000)); //todo: 通过kafka获取数据源获取数据模型进行数据处理 HashMapString, Object map new HashMap(); //配置kafka的消费组配置 map.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,192.168.1.189:9092); map.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer); map.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer); map.put(ConsumerConfig.GROUP_ID_CONFIG,heihei); map.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,latest); // 需要消费的主题 ArrayListString strings new ArrayList(); strings.add(sparkstreamingtopic); JavaInputDStreamConsumerRecordString, String directStream KafkaUtils .createDirectStream(jsc, LocationStrategies.PreferBrokers(), ConsumerStrategies.String, StringSubscribe(strings,map)); directStream.map(new FunctionConsumerRecordString, String, String() { Override public String call(ConsumerRecordString, String v1) throws Exception { return v1.value(); } }).print(100); jsc.start(); //阻塞代码他会一直阻塞在这里 jsc.awaitTermination(); }