题目描述 /** * 用户的行为日志数据清洗过滤 * 网站的独立访客数：一个唯一的IP地址就是一个独立访客 * 1、将用户行为日志数据中的IP地址字段获取到返回一个只包含IP地址的RDD * 2、RDD中重复的IP去重 * 3、RDD中的累加值 */ 案例 object A2DataAnaly { d
从集合（内存中创建出来）RDDval sparkConf = new sparkConf().setMaster("local[*]").setAppName("spark") val sparkContext = new SparkContext(sparkConf) val rdd1 = sparkContext.parallelize( List(1,2,3,4) ) val add2 = sparkContext.makeRDD( List(1,2,
RDD(2) RDD转换算子 RDD根据数据处理方式的不同将算子整体上分为Value类型、双Value类型、Key-Value类型 value类型 map 函数签名 def map[U:ClassTag](f:T=>U):RDD[U] 函数说明 将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换 e.g.1 val source = spa
(1)构建Spark Application的运行环境，启动SparkContext (2)SparkContext向资源管理器注册并向资源管理器申请运行Executor (3)资源管理器分配Executor并启动Executor (4)Executor发送心跳至资源管理器 (5)SparkContext构建DAG图 (6)将DAG分解成Stage,把Stage发送给taskScheduler (7
1、文件读取与保存 1.1、Text 文件 1）数据读取：textFile(String) 2）数据保存：saveAsTextFile(String) def main(args: Array[String]): Unit = { //1.创建SparkConf并设置App名称 val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("
/* * TODO 关于 SparkConf 的作用 * 1. spark的配置对象 用来初始化 Spark application 的配置信息 * 2. 用 SparkConf.set("key","value") 的方式来注入配置信息 * 3. 用 SparkConf对象 指定的配置信息,优先级是最高的(该对象的任何设置都会覆盖默认配置和系统属性) *
设置快捷键的原因 近期在梳理Spark ，为了便于获取上下文对象：即以下代码 //获取 SparkConf 并设置应用名称*本地模式 val conf: SparkConf = new SparkConf().setAppName("Spark").setMaster("local[8]") //获取 Spark 上下文对象 val sc: SparkContext = new Spar
目录结论SparkContext第三方集群模式 结论 SparkContext启动，根据setMaster设置的Url匹配运行模式。 SparkContext // 根据设置的master字符串的内容进行匹配 master match { // local模式 case "local" => ... ... // local[n] 和 local[*] 模式 n为threa
目录版本结论代码入口SparkContext -> textFileSparkContext -> defaultMinPartitionsSparkContext -> defaultParallelismTaskSchedulerImplLocalSchedulerBackend 版本 本分析基于Spark version 3.1.2 结论 local模式下，默认最小分区数不会超过2 如果对spark.default.parallelis
1.定义 /* * 1.定义 * def take(num: Int): Array[T] * 2.功能 * 返回一个由 RDD 的前 n 个元素组成的数组 * * */ 2.示例 object takeTest extends App { val sparkconf: SparkConf = new SparkConf().setMaster("local").setAppName("distinctTest")
package com.atguigu.bigata.spark.core.rdd.builder.operator.action import org.apache.spark.{SparkConf, SparkContext} import scala.collection.mutable /** * @auther :atom * @date :2022/2/20 20:40 * wordCount的九种写法 */ object Spark01_RDD_Operator_Action
今天是大年初三，猴赛雷 小谈 这几天每天晚上给她发一个红包，拜年红包而且还可以添加表情包。感觉现在过年好没有年味吖。嗑瓜子磕的嗓子都疼了。 Spark中的算子有很多，有Value类型，双Value类型，这两天写的都是Value类型的，昨天讲的是关于map的映射
Spark简介 Spark Streaming基于Spark Core实现了可扩展、高吞吐和容错的实时数据流处理。现在支持的数据源有Kafka、Flume、Twitter、ZeroMQ、Kinesis、HDFS、S3和TCP socket。处理后的结果可以存储到HDFS、Database或者Dashboard中。 Spark Streaming是将流式计算分解成一系列短
注意setMaster("local")和setMaster("local[*])的分区数是不一样的。*会匹配所有的cpu核数。 import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object Seq_Partition_04 { def main(args: Array[String]): Unit = { val sparkConf
今天继续学习sparkRDD的算子 （1）flatMap def main(args: Array[String]): Unit = { val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator11") val sparkContext = new SparkContext(sparkConf) val rdd: RDD[List[Int]]= sparkContext
今天接着昨天spark学习 使用spark自己自带的函数完成对文本中的单词统计，昨天代码的进阶版 代码：package com.chenghaixiang.core.wcimport org.apache.spark.rdd.RDDimport org.apache.spark.{SparkConf, SparkContext}object spark02_wordcount { def main(args: Array[String]):
累加器Accumulators 一，累加器作用及其原理1.1，作用1.2，原理 二，累加器关键源码跟踪阅读2.1，测试代码2.2，跟踪源码2.2.1，add调用2.2.2，merge调用 三，累加器在行动算子和转换算子中执行有何不同3.1，测试代码 四，累加器级别4.1,检查点与累加器4.2 多个行动操作与累加器 五，自定义累加器
Spark RDD-转换算子RDD-转换算子-mapRDD-转换算子-map-并行计算效果演示 RDD-转换算子-mapPartitionsRDD-转换算子-mapPartitions-小练习RDD-转换算子-mapPartitions&map的区别 RDD-转换算子-mapPartitionsWithIndexRDD-转换算子-flatMapRDD-转换算子-flatMap-小练习 RDD-
Spark实现WordCount的N种方法 大家好啊，这里就不自我介绍了，我们说一下WordCount，也就是词频。大家可能在各种渠道学习数据处理都会是WordCount首当其冲，为什么呢？因为WordCount简单。但是可以很好的形容数据处理和数据统计。今天我们也跟风的讲一讲WordCount，但是呢？我们
历经九九八十一难终于把IDEA的Scala开发环境配好了 /** * @author Jack Sun * @data 2021/11/19 - 20:00 */ //注意要想打个main就出来全部，必须要添加下面三个包 import org.apache.spark.SparkContext import org.apache.spark.SparkContext._ import org.apache.spark.Spark
1、定义 /** * Main entry point for Spark functionality. A SparkContext represents the connection to a Spark * cluster, and can be used to create RDDs, accumulators and broadcast variables on that cluster. * * Only one SparkContext may be active per JVM
本地模式 import org.apache.spark.SparkContext import org.apache.spark.SparkConf object WordCount { def main(args: Array[String]) { val conf = new SparkConf().setAppName("WordCount").setMaster("local[2]") val sc = new Spar
stypora-copy-images-to: img typora-root-url: ./ Spark Day02：Spark 基础环境（二） Hadoop3.0-HDFS https://www.bilibili.com/video/BV1yX4y1K7Lq Hadoop3.0-MapReduce https://www.bilibili.com/video/BV1Tf4y167U8 Hadoop3.0-yarn https://www.bilibili.com/video/BV1
package sparkcoreimport org.apache.spark.rdd.RDDimport org.apache.spark.{SparkConf, SparkContext}object Demo09Union { def main(args: Array[String]): Unit = { val conf: SparkConf = new SparkConf().setAppName("Union").setMaster("local&quo
释义 reduce将partition内所有记录最终计算成1个记录，结果类型与reduce 时数据类型一致 方法签名如下： def reduce(f: (T, T) => T): T = withScope { ... } f: 每个partition内，每每2个记录进行归并计算。输入类型为U跟U，输出为U，即输入类型与输出类型一致 案例 计算数组总和
