从集合（内存中创建出来）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,
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
目录结论代码SparkConf -> setAppNameSparkConf -> set 重载方法(两个参数)SparkConf 结论 SparkConf设置的key / value 保存在SparkConf名为settings的变量中，该变量类型为ConcurrentHashMap SparkConf的setXX方法, 封装了特定的属性名和值存储到map中 SparkConf的set方法，直接将
本地Spark程序调试需要使用local提交模式，即将本机当做运行环境，Master和Worker都为本机。运行时直接加断点调试即可。如下： 创建SparkConf的时候设置额外属性，表明本地执行： val conf = new SparkConf().setAppName("WC").setMaster("local[*]") 如果本机操作系统是windows，如
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")
1、安装pyarrow加速 pyspark 2.3 对应pyarrow的版本是0.14.1 2、pycharm需要配置的环境量 HADOOP_HOME /opt/hdp/2.3.4.0-315/hadoop SPARK_HOME /opt/hdp/2.3.4.0-315/spark2 PYTHONPATH /data/soft/anaconda3/envs/py37/bin/python PYSPARK_PYTHON /data/soft/anaconda3/e
今天学习spark的一些行动算子和序列化 （1）行动算子 def main(args: Array[String]): Unit = { val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator") val sc = new SparkContext(sparkConf) val rdd = sc.makeRDD(List(("a",1),(&quo
注意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
今天继续学习spark双value算子 （1） def main(args: Array[String]): Unit = { val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator15") val sc = new SparkContext(sparkConf) //双value，数据源类型要保持一致，拉链类型可以不一致 //会报错，因为拉链分区
今天学习完单value的算子和双value算子的开始 （1）distinct def main(args: Array[String]): Unit = { val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator15") val sc = new SparkContext(sparkConf) //distinct,去重 val rdd=sc.makeRDD(Lis
今天继续学习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
1、实验要求 对给定的一个英文文本，使用Spark完成文本内容的读取并转换成RDD，然后使用RDD的算子统计每个单词出现的次数，将统计结果按从大到小的顺序打印到控制台上。 2、实验代码 import findspark findspark.init() from pyspark import SparkContext, SparkConf sparkConf = Spar
1、所有的配置存储在一个ConcurrentHashMap中，名为settings。 2、Spark的参数配置有以下三种获取。 源于系统参数中以spark为前缀的属性。（通过System.getProperties获取） 使用SparkConf的API设置的。 从其他SparkConf中克隆
package sparkcoreimport org.apache.spark.rdd.RDDimport org.apache.spark.{SparkConf, SparkContext}object Demo04FlatMap { def main(args: Array[String]): Unit = { val conf: SparkConf = new SparkConf().setAppName("FlatMap").setMaster("local
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
使用IDEA工具运行Spark的WordCount时，运行报错，报错信息如下 Exception in thread "main" java.lang.NoSuchMethodError: scala.Product.$init$(Lscala/Product;)V at org.apache.spark.SparkConf$DeprecatedConfig.<init>(SparkConf.scala:810) at org.apache.spark.SparkConf
文章目录保存为sequenceFile保存到HDFS保存到mysql保存为sequenceFilepackage write import org.apache.hadoop.io.compress.GzipCodec import org.apache.spark.{SparkConf, SparkContext} object saveToSeq { def main(args: Array[String]): Unit = {
Spark的数据读取及数据保存可以从两个维度来作区分：文件格式以及文件系统。 文件格式分为：Text文件、Json文件、Csv文件、Sequence文件以及Object文件 文件系统分为：本地文件系统、HDFS以及数据库 一、文件类数据读取与保存 1.1 Text文件 数据读取：textFile(String)数据保存：save
Value类型-groupBy()分组 1、groupBy()算子函数签名： def groupBy(f : T =>K]) : RDD[K, Iterable[T]] 2、算子·功能： 1、将原来RDD中的元素按照groupBy中的函数f的执行结果进行分组 2、结果相同的元素进入同一个迭代器 3、groupBy算子存在shuffle 4、shuffle会落盘，元素会在不
通过集合创建RDD val conf: SparkConf = new SparkConf().setAppName(this.getClass.getName).setMaster("local[*]") val sc = new SparkContext(conf) // 方式一：parallelize方法创建RDD //val rdd: RDD[Int] = sc.parallelize(List(1,2,3,4)) // 方式二：makeRDD方法创建RD
1.什么是序列化 我们知道内存中的数据对象只有转化成二级制的流才可以进行数据的持久化和网络传输，序列化是将数据对象转换为字节序列的过程，而反序列化是指把字节序列恢复为数据对象的过程。 序列化需要保留充分的信息以恢复数据对象，但是为了节约存储空间和网
数据中有设备的识别码 一共有7个标签 里面涉及到词库 所以要用到ES会更加合理 这写标签都是围绕用户的id，一共会有15种id 在用户标签下，如果id用户情况相同时，相同标签的累加（求交集），不同的求并集 上下文标签的实现 在一天的时间内 早上产生可一个日志 晚上也产生一条数
Spark Streaming 是微批处理。 SparkConf sparkConf = new SparkConf().setAppName("SparkStreaming").setMaster("local[*]"); JavaStreamingContext javaStreamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(1000)); Durations.sec
