SparkStreaming+Kafka 实现基于缓存的实时wordcount程序

2018-06-14

前言

本文利用SparkStreaming和Kafka实现基于缓存的实时wordcount程序，什么意思呢，因为一般的SparkStreaming的wordcount程序比如官网上的，只能统计最新时间间隔内的每个单词的数量，而不能将历史的累加起来，本文是看了教程之后，自己实现了一下kafka的程序，记录在这里。其实没什么难度，只是用了一个updateStateByKey算子就能实现，因为第一次用这个算子，所以正好学习一下。

1、数据

数据是我随机在kafka里生产的几条，单词以空格区分开

2、kafka topic

首先在kafka建一个程序用到topic:UpdateStateBykeyWordCount

1	bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic UpdateStateBykeyWordCount

3、创建checkpoint的hdfs目录

我的目录为：/spark/dkl/kafka/wordcount_checkpoint

1	hadoop fs -mkdir -p /spark/dkl/kafka/wordcount_checkpoint

more >>

展开全文 >>

Spark架构原理

2018-06-09

前言

本文总结了Spark架构原理,其中主要包括五个组件：Driver、Master、Worker、Executor和Task，简要概括了每个组件是干啥的，并总结提交spark程序之后，这五个组件运行的详细步骤。

1、流程图

为了直观，就把流程图放在最前面了

more >>

展开全文 >>

Spark 持久化（cache和persist的区别）

2018-06-03

1、RDD 持久化

Spark 中一个很重要的能力是将数据持久化（或称为缓存），在多个操作间都可以访问这些持久化的数据。当持久化一个 RDD 时，每个节点的其它分区都可以使用 RDD 在内存中进行计算，在该数据上的其他 action 操作将直接使用内存中的数据。这样会让以后的 action 操作计算速度加快（通常运行速度会加速 10 倍）。缓存是迭代算法和快速的交互式使用的重要工具。

RDD 可以使用 persist() 方法或 cache() 方法进行持久化。数据将会在第一次 action 操作时进行计算，并缓存在节点的内存中。Spark 的缓存具有容错机制，如果一个缓存的 RDD 的某个分区丢失了，Spark 将按照原来的计算过程，自动重新计算并进行缓存。

在 shuffle 操作中（例如 reduceByKey），即便是用户没有调用 persist 方法，Spark 也会自动缓存部分中间数据。这么做的目的是，在 shuffle 的过程中某个节点运行失败时，不需要重新计算所有的输入数据。如果用户想多次使用某个 RDD，强烈推荐在该 RDD 上调用 persist 方法。

2、存储级别

每个持久化的 RDD 可以使用不同的存储级别进行缓存，例如，持久化到磁盘、已序列化的 Java 对象形式持久化到内存（可以节省空间）、跨节点间复制、以 off-heap 的方式存储在 Tachyon。这些存储级别通过传递一个 StorageLevel 对象给 persist() 方法进行设置。
详细的存储级别介绍如下：

MEMORY_ONLY : 将 RDD 以反序列化 Java 对象的形式存储在 JVM 中。如果内存空间不够，部分数据分区将不再缓存，在每次需要用到这些数据时重新进行计算。这是默认的级别。
MEMORY_AND_DISK : 将 RDD 以反序列化 Java 对象的形式存储在 JVM 中。如果内存空间不够，将未缓存的数据分区存储到磁盘，在需要使用这些分区时从磁盘读取。
MEMORY_ONLY_SER : 将 RDD 以序列化的 Java 对象的形式进行存储（每个分区为一个 byte 数组）。这种方式会比反序列化对象的方式节省很多空间，尤其是在使用 fast serializer时会节省更多的空间，但是在读取时会增加 CPU 的计算负担。
MEMORY_AND_DISK_SER : 类似于 MEMORY_ONLY_SER ，但是溢出的分区会存储到磁盘，而不是在用到它们时重新计算。
DISK_ONLY : 只在磁盘上缓存 RDD。
MEMORY_ONLY_2，MEMORY_AND_DISK_2，等等 : 与上面的级别功能相同，只不过每个分区在集群中两个节点上建立副本。
OFF_HEAP（实验中）: 类似于 MEMORY_ONLY_SER ，但是将数据存储在 off-heap memory，这需要启动 off-heap 内存。

注意，在 Python 中，缓存的对象总是使用 Pickle 进行序列化，所以在 Python 中不关心你选择的是哪一种序列化级别。python 中的存储级别包括 MEMORY_ONLY，MEMORY_ONLY_2，MEMORY_AND_DISK，MEMORY_AND_DISK_2，DISK_ONLY 和 DISK_ONLY_2 。
more >>

展开全文 >>

Scala日期操作

2018-06-01

前言

本文主要记录我自己对日期格式数据的一些常用操作，主要目的是备忘，方便随时查阅。本文没有将代码封装为函数，如果有需要的可以自行封装，注意每一部分的代码会依赖前面代码里的变量。

代码可以直接在spark-shell里运行（在scala里有的包没有）

1、字符串转日期

import java.text.SimpleDateFormat
import org.joda.time.DateTime
val dateStr = "2018-06-01"
val pattern = "yyyy-MM-dd"
val date = new SimpleDateFormat(pattern).parse(dateStr)
val dateTime = new DateTime(date)
println(date)
println(dateTime)

1 2	Fri Jun 01 00:00:00 CST 2018 2018-06-01T00:00:00.000+08:00

2、日期转字符串

将上面的日期转成其他格式的字符串

1	println(new SimpleDateFormat("yyyyMMdd").format(date))

20180601

more >>

展开全文 >>

Spark读取压缩文件

2018-05-30

前言

本文讲如何用spark读取gz类型的压缩文件，以及如何解决我遇到的各种问题。

1、文件压缩

下面这一部分摘自Spark快速大数据分析：
在大数据工作中，我们经常需要对数据进行压缩以节省存储空间和网络传输开销。对于大多数Hadoop输出格式来说，我们可以指定一种压缩编解码器来压缩数据。
选择一个输出压缩编解码器可能会对这些数据以后的用户产生巨大影响。对于像Spark 这样的分布式系统，我们通常会尝试从多个不同机器上一起读入数据。要实现这种情况，每个工作节点都必须能够找到一条新记录的开端。有些压缩格式会使这变得不可能，而必须要单个节点来读入所有数据，这就很容易产生性能瓶颈。可以很容易地从多个节点上并行读取的格式被称为“可分割”的格式。下表列出了可用的压缩选项。

格式	可分割	平均压缩速度	文本文件压缩效率	Hadoop压缩编解码器	纯Java实现	原生	备注
gzip	否	快	高	org.apache.hadoop.io.compress.GzipCodec	是	是
lzo	是（取决于所使用的库）	非常快	中等	com.hadoop.compression.lzo.LzoCodec	是	是	需要在每个节点上安装LZO
bzip2	是	慢	非常高	org.apache.hadoop.io.compress.Bzip2Codec	是	是	为可分割版本使用纯Java
zlib	否	慢	中等	org.apache.hadoop.io.compress.DefaultCodec	是	是	Hadoop 的默认压缩编解码器
Snappy	否	非常快	低	org.apache.hadoop.io.compress.SnappyCodec	否	是	Snappy 有纯Java的移植版，但是在Spark/Hadoop中不能用

尽管Spark 的textFile() 方法可以处理压缩过的输入，但即使输入数据被以可分割读取的方式压缩，Spark 也不会打开splittable。因此，如果你要读取单个压缩过的输入，最好不要考虑使用Spark 的封装，而是使用newAPIHadoopFile 或者hadoopFile，并指定正确的压缩编解码器。

关于上面一段话的个人测试：选取一个大文件txt，大小为1.5G，写spark程序读取hdfs上的该文件然后写入hive，经测试在多个分区的情况下，txt执行时间最短，因为在多个机器并行执行，而gz文件是不可分割的，即使指定分区数目，但依然是一个分区，一个task，即在一个机器上执行，bzip2格式的文件虽然是可分割的，即可以按照指定的分区分为不同的task在多个机器上执行，但是执行时间长，比gz时间还长，经过四次改变bzip2的分区，发现最快的时间和gz时间是一样的，如果指定一个分区的话，比gz要慢很多，我想这样就可以更好的理解：”尽管Spark 的textFile() 方法可以处理压缩过的输入，但即使输入数据被以可分割读取的方式压缩，Spark 也不会打开splittable”这句话了。

more >>

展开全文 >>