小雷のBlog - 大数据
type: channel
link: https://blog.xzlei.cn/category/bigdata/
description: bigdata
items: Spark快速入门
items: Flink快速入门

Spark快速入门
type: item
date: 2026-01-05T19:26:00+08:00
description: AI摘要 本文介绍了Apache Spark的快速入门，包括Maven依赖配置、最简单的入门程序、RDD转换算子、行动算子以及累加器和广播变量的使用。通过具体代码示例，详细解释了Spark的基本概念和操作，帮助初学者快速掌握Spark的基本使用方法。 此内容根据文章生成，仅用于文章内容的解释与总结 Spark maven依赖配置<?xml version="1.0" encoding="UTF-8"?> 4.0.0 cn.xzlei featureEngineering 1.0-SNAPSHOT <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> org.apache.spark spark-core_2.12 3.0.0 net.alchim31.maven scala-maven-plugin 3.2.2 org.apache.maven.plugins maven-assembly-plugin 3.1.0 jar-with-dependencies 最简单的入门程序请先创建一个项目，然后导入相应依赖，在项目根目录下创建一个datas文件夹用来存放数据文件，在datas目录下创建一个1.txt，输入如下数据：hello word hello scala hello sparkpackage cn.xzlei.test import org.apache.spark.{SparkConf, SparkContext} object Test { def main(args: Array[String]): Unit = { // 创建Spark配置项 val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Test") // 根据配置项创建Spark上下文执行环境 val sc = new SparkContext(sparkConf) // 读取文本文件datas/1.txt val wordText = sc.textFile("datas/1.txt") // 根据flatMap扁平化操作根据空格进行单词分割，("hello","word","hello") val words = wordText.flatMap(_.split(" ")) // 调用map函数进行映射，转为（String,Int）类型：(hello,1) val wordByOne = words.map(item => (item, 1)) // 调用reduceByKey方法进行聚合操作 wordByOne.reduceByKey(_+_).foreach(println) sc.stop() } } 运行效果如下：RDD转换算子RDD根据数据处理方式的不同将算子整体上分为Value类型、双Value类型和Key-Value类型Value类型map函数签名：def map[U: ClassTag](f: T => U): RDD[U]函数说明：将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换。示例代码：val dataRDD: RDD[Int] = sparkContext.makeRDD(List(1,2,3,4)) val dataRDD1: RDD[Int] = dataRDD.map(num => { num * 2 }) val dataRDD2: RDD[String] = dataRDD1.map(num => { "" + num })flatMap函数签名：def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]函数说明：将处理的数据进行扁平化后再进行映射处理，所以算子也称之为扁平映射示例代码：val dataRDD = sparkContext.makeRDD(List( List(1,2),List(3,4) ),1) val dataRDD1 = dataRDD.flatMap( list => list )groupBy函数签名：def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]函数说明：将数据根据指定的规则进行分组, 分区默认不变，但是数据会被打乱重新组合，我们将这样的操作称之为sh…

Flink快速入门
type: item
date: 2026-01-02T16:16:00+08:00
description: AI摘要 本文介绍了Flink的快速入门方法，包括如何启动集群、提交任务、查看任务信息，以及如何读取有界数据和Kafka数据。文章提供了详细的命令行操作步骤和示例代码，帮助读者快速掌握Flink的基本使用方法。 此内容根据文章生成，仅用于文章内容的解释与总结 {music-list id="3136952023" color="#1989fa" autoplay="autoplay"/}Flink命令行提交任务启动集群bin/start-cluster.sh在 node1 中执行以下命令启动 netcat。nc -lk 7777进入到 Flink 的安装路径下，在命令行使用 flink run 命令提交作业。./bin/flink run -m node1:8081 -c cn.xzlei.chap02.StreamWordCount ./flumeDemo-1.0-SNAPSHOT.jar -param --host node1 --port 7777参数： –m 指定了提交到的 JobManager，-c 指定了入口类-param 指定了jar的自定义参数查看指定任务信息bin/flink-cluster.sh list [taskID]查看所有任务信息bin/flink-cluster.sh list读取有界数据package cn.xzlei.chap05 import org.apache.flink.streaming.api.scala._ import java.util.Date case class Event(user:String,url:String,timestamp:Long) object SourceBoundedTest { def main(args: Array[String]): Unit = { // 创建一个执行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment // 设置并行数量为1 env.setParallelism(1) // 创建一个数据列表 val list:List[Event] = List( Event("Piter", "./home?id=5", new Date().getTime), Event("Mary", "./shop?id=8", new Date().getTime), Event("Jack", "./userCenter?id=78", new Date().getTime), ) // 从集合中获取数据 val stream1 = env.fromCollection(list) // 从元素中获取数据 val stream2 = env.fromElements(1, 2, 3, 4, 5) // 从文本文件中获取数据 val stream3 = env.readTextFile("input/test.txt") stream1.print("stream1") stream2.print("number") stream3.print("stream3") env.execute() } } 读取Kafka数据添加pom.xml文件依赖： org.apache.flink flink-connector-kafka_2.12 1.14 编写代码package cn.xzlei.chap05 import org.apache.flink.api.common.eventtime.WatermarkStrategy import org.apache.flink.api.common.serialization.SimpleStringSchema import org.apache.flink.connector.kafka.source.KafkaSource import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer import org.apache.flink.streaming.api.scala._ import java.util.Properties object SourceKafkaTest { def main(args: Array[String]): Unit = { // 创建一个流处理环境 val env = StreamExecutionEnvironment.getExecutionEnvironment // 创建一个sourceKafka连接源 val kafkaSource = KafkaSource.…