Spark性能优化：基于分区进行操作

前言（摘自Spark快速大数据分析）

基于分区对数据进行操作可以让我们避免为每个数据元素进行重复的配置工作。诸如打开数据库连接或创建随机数生成器等操作，都是我们应当尽量避免为每个元素都配置一次的工作。Spark 提供基于分区的map 和foreach，让你的部分代码只对RDD 的每个分区运行一次，这样可以帮助降低这些操作的代价。
当基于分区操作RDD 时，Spark 会为函数提供该分区中的元素的迭代器。返回值方面，也返回一个迭代器。除mapPartitions() 外，Spark 还有一些别的基于分区的操作符，见下表：

函数名 调用所提供的 返回的 对于RDD[T]的函数签名
mapPartitions() 该分区中元素的迭代器 返回的元素的迭代器 f: (Iterator[T]) → Iterator[U]
mapPartitionsWithIndex() 分区序号，以及每个分区中的元素的迭代器 返回的元素的迭代器 f: (Int, Iterator[T]) → Iterator[U]
foreachPartitions() 元素迭代器 无 f: (Iterator[T]) → Unit
more >>

展开全文 >>

利用Spark实现Oracle到Hive的历史数据同步

1、需求背景

和上一篇文章Spark通过修改DataFrame的schema给表字段添加注释一样，通过Spark将关系型数据库（以Oracle为例）的表同步的Hive，这里讲的只是同步历史数据，不包括同步增量数据。

2、Oracle和Hive的字段类型对应

利用Spark的字段类型自动匹配，本来以为Spark匹配的不是很好，只是简单的判断一下是否为数字、字符串，结果经验证，Spark可以获取到Oracle的小数点精度，Spark的字段类型对应和我自己整理的差不多，所以就索性用Spark自带的字段类型匹配，而不是自己去Oracle相关表获取每个字段类型，然后一一转化为Hive对应的字段类型，下面是Oracle和Hive的字段类型对应，只是整理了大概：

Oracle Hive
VARCHAR2 String
NVARCHAR2 String
NUMBER DECIMAL/Int
DATE TIMESTAMP
TIMESTAMP TIMESTAMP
CHAR String
CLOB（一般用不到） String
BLOB（一般用不到） BINARY
RAW （一般用不到） BINARY
Other String
more >>

展开全文 >>

Spark通过修改DataFrame的schema给表字段添加注释

1、需求背景

通过Spark将关系型数据库（以Oracle为例）的表同步的Hive表，要求用Spark建表，有字段注释的也要加上注释。Spark建表，有两种方法：

  • 用Spark Sql，在程序里组建表语句，然后用Spark.sql(“建表语句”)建表，这种方法麻烦的地方在于你要读取Oracle表的详细的表结构信息，且要进行Oracle和Hive的字段类型进行一一对应
  • 用DataFrame 的saveAsTable方法，这种方法如果对应的数据库里没有表，则Spark会根据DataFrame的schema自动建表，比较简单，不用考虑字段类型匹配转化问题，但是这种方法有一个问题，Spark读取Oracle的表为DataFrame时，并不能将表字段的注释读进来，所以就有了如标题所示的需求。（一开始以为DataFrame不能加注释，经过研究，发现是可以的！）

    2、如何查看DataFrame是否有注释

前面讲到DataFrame里没有Oracle的注释信息，但是如果数据源为Hive的话，是可以将注释获取到的。

2.1 新建Hive测试表（带注释）

1
2
3
4
5
create table `test` (
`id` string comment 'ID', 
`Name` string comment '名字'
)
comment '测试';
more >>

展开全文 >>

Spark创建空的DataFrame

前言

本文主要给出Spark创建空的DataFrame的代码示例，这里讲的空的DataFrame主要指有列名（可以自己随意指定），但是没有行的DataFrame，因为自己在开发过程中有这个需求，之前并不知道怎么创建，就查了一下，发现资料并不多，不知道因为太简单还是用的人少~，至于具体什么需求就不阐述了，主要给有这方面需求的小伙伴参考一下。还有另一种空的DataFrame就是没有任何行任何列的DataFrame，不知道有什么用，反正贴在代码里，万一有人用呢~

1、代码

代码较简单，如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
package com.dkl.leanring.spark.df

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
import org.apache.spark.rdd.EmptyRDD

/**
 * Spark创建空DataFrame示例
 */
object EmptyDataFrame {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().appName("EmptyDataFrame").master("local").getOrCreate()

    /**
     * 创建一个空的DataFrame，代表用户
     * 有四列，分别代表ID、名字、年龄、生日
     */
    val colNames = Array("id", "name", "age", "birth")
    //为了简单起见，字段类型都为String
    val schema = StructType(colNames.map(fieldName => StructField(fieldName, StringType, true)))
    //主要是利用了spark.sparkContext.emptyRDD
    val emptyDf = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], schema)

    emptyDf.show

    /**
     * 也可以给每列指定相对应的类型
     */
    val schema1 = StructType(
      Seq(
        StructField("id", IntegerType, true),
        StructField("name", StringType, true),
        StructField("age", IntegerType, true),
        StructField("birth", StringType, true)))
    val emptyDf1 = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], schema1)
    emptyDf1.show

    //还有一种空的DataFrame，没有任何行任何列
    spark.emptyDataFrame.show

    spark.stop()
  }

}

more >>

展开全文 >>

Spark 创建RDD、DataFrame各种情况的默认分区数

前言

熟悉Spark的分区对于Spark性能调优很重要，本文总结Spark通过各种函数创建RDD、DataFrame时默认的分区数，其中主要和sc.defaultParallelism、sc.defaultMinPartitions以及HDFS文件的Block数量有关，还有很坑的某些情况的默认分区数为1。

  • 如果分区数少，那么并行执行的task就少，特别情况下，分区数为1，即使你分配的Executor很多，而实际执行的Executor只有1个，如果数据很大的话，那么任务执行的就很慢，好像是卡死了~，所以熟悉各种情况下默认的分区数对于Spark调优就很有必要了，特别是执行完算子返回的结果分区数为1的情况，更需要特别注意。（我就被坑过，我已经分配了足够多的Executor、默认的并行度、以及执行之前的数据集分区数，但分区数依然为1）

1、关于 sc.defaultMinPartitions

sc.defaultMinPartitions=min(sc.defaultParallelism,2)
也就是sc.defaultMinPartitions只有两个值1和2，当sc.defaultParallelism>1时值为2，当sc.defaultParallelism=1时，值为1
上面的公式是在源码里定义的（均在类SparkContext里）：

1
2
3
4
5
def defaultMinPartitions: Int = math.min(defaultParallelism, 2)
def defaultParallelism: Int = {
    assertNotStopped()
    taskScheduler.defaultParallelism
  }

2、关于sc.defaultParallelism

2.1 首先可通过spark.default.parallelism设置sc.defaultParallelism的值

more >>

展开全文 >>

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

2016年本科毕业于合肥工业大学<br><br>信息与计算科学专业<br><br>目前在济南一家公司从事大数据的工作~~