Spark读写压缩文件

spark 自动根据文春亩件后缀名判断压缩格式，不用特别指定

spark会加载Hadoop的默认的配置，如果hadoop中设置了压缩，spark没指定压缩则会用hadoop的配置方式压缩；

spark中指定方式斗销如下：

方法1：写文件时指定压缩格式空森游

方法2：配置方式，全局可用

如果数据量很少不需要压缩只需设置不压缩： sc.hadoopConfiguration.set(FileOutputFormat.COMPRESS , "false")

spark进入txt文件的命令

1、首先启动spark-shell进入Spark-shell模式：（进入spark目录下后 输入命令 bin/spark-shell启动spark-shell模式）

2、加载text文件（spark创建sc，可以加载本地文件和HDFS文件戚庆州创建RDD）

val textFile = sc.textFile("file:///home/hadoop/test1.txt") #注意file:后是三个“/”

注意：加载HDFS文件和本地文件都是使用textFile，区别是添加前缀(hdfs://和file://)进行标识。

3、获取RDD文件textFile所有项（文本文件即总共行数）的计数（还有很多其他的RDD *** 作，自行百度）

textFile.count() #统计结果显示 1 行

二、在 spark-shell 中读取 HDFS 系统文件“/home/hadoop/test.csv（也可以是txt文件）”（如果该差告文件不存在， 请先创建），然后，统计出文件的行数：

方法一：

1、加载text文件（spark创建sc，可以加载本地文件和HDFS文件创建RDD）

val textFile = sc.textFile("hdfs:///home/hadoop/test.csv") #注意hdfs:后是三个“/”

注意：加载HDFS文件和本地文件都是使用textFile，区别是添加前缀(hdfs://和file://)进行标识。

2、获取RDD文件textFile所有项的计数

textFile.count() #统计结果显示 1 行

方法二：（Spark shell 默认是读取 HDFS 中的文件，需要先上传文件到 HDFS 中，否则会有“org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: hdfs://localhost:9000/user/hadoop/README.md”的错误。）

1、省去方法一中第一步的命令（1）中的“hdfs：//”，其他部分相同，命令如下：

三、编写独立应用程序，读取 HDFS 系统文件“/user/hadoop/test.txt”（如果该文件不存在， 请先创建），然后，统计出文件的行数；通过 sbt 工具将整个应用程序编译打包成 JAR 包， 并将生成的 JAR 包通过 spark-submit 提交到 Spark 中运行命令：

1、首先输入:quit 命令退出spark-shell模式：

2、在终端中执行如下命令创建一个文件夹 sparkapp3 作为应用程序根目录：

cd ~ # 进入用户主文件夹

mkdir ./sparkapp3 # 创建应用程序根目录

mkdir -p ./sparkapp3/src/高蔽main/scala # 创建所需的文件夹结构

3、在 ./sparkapp3/src/main/scala 下建立一个名为 SimpleApp.scala 的文件（vim ./sparkapp3/src/main/scala/SimpleApp.scala），添加代码如下：

/* SimpleApp.scala */

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

import org.apache.spark.SparkConf

object SimpleApp {

def main(args: Array[String]) {

val logFile = "hdfs://localhost:9000/home/hadoop/test.csv"

val conf = new SparkConf().setAppName("Simple Application")

val sc = new SparkContext(conf)

val logData = sc.textFile(logFile, 2)

val num = logData.count()

println("这个文件有 %d 行！".format(num))

}

}

4、该程序依赖 Spark API，因此我们需要通过 sbt 进行编译打包。 ./sparkapp3 中新建文件 simple.sbt（vim ./sparkapp3/simple.sbt），添加内容如下，声明该独立应用程序的信息以及与 Spark 的依赖关系：

name := "Simple Project"

version := "1.0"

scalaVersion := "2.12.10"

libraryDependencies += "org.apache.spark" %% "spark-core" % "3.0.0-preview2"

elasticsearch-spark 提供了saveToEs api以支持快速导入数据。但es集群线程池有限，在大量写入数据的同时，对cpu的压力非常大，影响线上es的查询服务。如果能参考hbase 的bulkload方法，对es也采用“bulkload”模式，写入性能会有巨大提升。核心思想是通过spark作业生成es的lucene文件，并通过网络传输，写入es的数据文件。明脊

本方案参考滴滴的fastIndex： 滴拆迅滴FastIndex

采用spark改写，部分特性适应了公司的原始流程，会有不一样的地方。如您采用的是spark saveToEs需要通过旅槐此该方法进行改写，可参考。

git地址为： https://github.com/Dengyu123/fast-es-rdd

---