❶ spark中怎么将读取的每一行的数据按某几行拼接成一行 新手，求指教，谢谢！
assert(args.length>1)
val_from=args(0)
val_to=args(1)
vals=sc.textFile(_from).collect()
valn=if(args.length>2)args(2).toIntelse2
valnumSlices=s.length/n
valx=sc.parallelize(s,numSlices).zipWithIndex().aggregate(List.empty[List[String]])(seqOp=(result,lineWithIndex)=>{
lineWithIndexmatch{
case(line,index)=>
if(index%n==0)List(line)::resultelse(line::result.head)::result.tail
}
},combOp=(x,y)=>x:::y).map(_.reversemkString"")
sc.parallelize(x).saveAsTextFile(_to)
textFile 不能指定分区数目，所以只能parallelize， n是每几行一合并，RDD的aggregate方法与foldLeft类似，因为并行，合并之后的行间顺序不确定的
下面给出非RDD操作示例
vals=List("123","234","345","456","567","678")
valn=3
//sparkRDD没有foldLeft
valx=s.zipWithIndex.foldLeft(List.empty[List[String]])((result,lineWithIndex)=>{
lineWithIndexmatch{
case(line,index)=>
if(index%n==0)List(line)::resultelse(line::result.head)::result.tail
}
}).reverse.map(_.reverse)
println(x)
❷ 用spark获取日志文件中记录内容
使用Apache Spark可以方便地读取并处理日志文件中的记录内容。
下面是一个使用Spark读取日志文件中的记录内容的示例代码：
# 导入Spark相关的库
from pyspark import SparkContext, SparkConf
# 创建SparkContext对象
sc = SparkContext(appName="Log Processing")
# 读取日志文件
log_file = sc.textFile("/path/to/log/file.log")
# 按行解析日志记录
log_records = log_file.map(lambda line: line.split(" "))
# 过滤出指定类型的日志记录
filtered_records = log_records.filter(lambda record: record[2] == "ERROR")
# 对日志记录进行处理，如统计数量或分析日志信息等
# ...
# 关闭SparkContext
sc.stop()
上面的示例代码首先使用Spark的textFile()方法读取日志文件，然后使用map()方法将日志文件的每一行按空格分割成一个数组，得到一个日志记录的RDD。接着使用filter()方法过滤出指定类型的日志记录，最后对日志记录进行处理。
使用Spark处理日志文件的优点在于，可以利用Spark的分布式计算能力，对大量的日志文件进行快速的处理。例如，可以使用Spark的MapRece算法快速统计日志文件中各种类型的记录数量，或者使用Spark SQL快速查询日志文件中的特定信息。
此外，Spark还提供了丰富的API和算法库，可以方便地进行数据清洗、数据分析和特征提取等复杂的数据处理任务。例如，可以使用Spark MLlib库进行机器学习，或者使用Spark GraphX库进行图计算等。
总之，使用Spark可以方便地读取并处理日志文件中的记录内容，是一种高效的数据处理方式。
❸ spark与hdfs怎么加载本地文件
hdfs和本地文件系统是并列的， 都可以使用spark进行加载， 只不过 hdfs是分布式的文件系统， spark 可以使用 sc.textFile("hdfs://....")操作hdfs的文件也可以 sc.textFile("本地文件路径")来加载本地文件的路径。
❹ Spark 怎么读文件名
java">//多易大数据
importorg.apache.hadoop.io.{LongWritable,Text}
importorg.apache.hadoop.mapred.{FileSplit,InputSplit,TextInputFormat}
importorg.apache.spark.{SparkConf,SparkContext}
importorg.apache.spark.rdd.HadoopRDDobjecttestPathOld{
defmain(args:Array[String]):Unit={
valconf=newSparkConf()
conf.setAppName("testtoParquet")
conf.setMaster("local")
valsc=newSparkContext(conf)
varinput="/home/dwj/data/testSpark/20180409"
valfileRDD=sc.hadoopFile[LongWritable,Text,TextInputFormat](input)
valhadoopRDD=fileRDD.asInstanceOf[HadoopRDD[LongWritable,Text]]
valfileAdnLine=hadoopRDD.mapPartitionsWithInputSplit((inputSplit:InputSplit,iterator:Iterator[(LongWritable,Text)])=>{
valfile=inputSplit.asInstanceOf[FileSplit]
iterator.map(x=>{file.getPath.toString()+" "+x._2})
})
fileAdnLine.foreach(println)
}
}
❺ spark saveAsTextfile 方法保存的文件part-00000 是空文件
spark中saveAsTextFile如何最终生成一个文件
一般而言，saveAsTextFile会按照执行task的多少生成多少个文件，比如part-00一直到part-0n，n自然就是task的个数，亦即是最后的stage的分区数。那么有没有办法最后只生成一个文件，而不是成百上千个文件了？答案自然是有办法。
在RDD上调用coalesce(1,true).saveAsTextFile()，意味着做完计算之后将数据汇集到一个分区，然后再执行保存的动作，显然，一个分区，Spark自然只起一个task来执行保存的动作，也就只有一个文件产生了。又或者，可以调用repartition(1)，它其实是coalesce的一个包装，默认第二个参数为true。
事情最终就这么简单吗？显然不是。你虽然可以这么做，但代价是巨大的。因为Spark面对的是大量的数据，并且是并行执行的，如果强行要求最后
只有一个分区，必然导致大量的磁盘IO和网络IO产生，并且最终执行rece操作的节点的内存也会承受很大考验。Spark程序会很慢，甚至死掉。
这往往是初学Spark的一个思维陷阱，需要改变原先那种单线程单节点的思维，对程序的理解要转变多多个节点多个进程中去，需要熟悉多节点集群自然产生多个文件这种模式。
此外，saveAsTextFile要求保存的目录之前是没有的，否则会报错。所以，最好程序中保存前先判断一下目录是否存在。
❻ [spark] Shuffle Read解析 (Sort Based Shuffle)
Shuffle Write 请看 Shuffle Write解析 。
本文将讲解shuffle Rece部分，shuffle的下游Stage的第一个rdd是ShuffleRDD，通过其compute方法来获取上游Stage Shuffle Write溢写到磁盘文件数据的一个迭代器：
从SparkEnv中获取shuffleManager（这里是SortShuffleManager），通过manager获取Reader并调用其read方法来得到一个迭代器。
getReader方法实例化了一个BlockStoreShuffleReader，参数有需要获取分区对应的partitionId，看看起read方法：
首先实例化了ShuffleBlockFetcherIterator对象，其中一个参数：
该方法获取rece端数据的来源的元数据，返回的是 Seq[(BlockManagerId, Seq[(BlockId, Long)])]，即数据是来自于哪个节点的哪些block的，并且block的数据大小是多少，看看getMapSizesByExecutorId是怎么实现的：
跟进getStatuses：
若能从mapStatuses获取到则直接返回，若不能则向mapOutputTrackerMaster通信发送GetMapOutputStatuses消息来获取元数据。
我们知道一个Executor对应一个CoarseGrainedExecutorBackend，构建CoarseGrainedExecutorBackend的时候会创建一个SparkEnv，创建SparkEnv的时候会创建一个mapOutputTracker，即mapOutputTracker和Executor一一对应，也就是每一个Executor都有一个mapOutputTracker来维护元数据信息。
这里的mapStatuses就是mapOutputTracker保存元数据信息的，mapOutputTracker和Executor一一对应，在该Executor上完成的Shuffle Write的元数据信息都会保存在其mapStatus里面，另外通过远程获取的其他Executor上完成的Shuffle Write的元数据信息也会在当前的mapStatuses中保存。
Executor对应的是mapOutputTrackerWorker，而Driver对应的是mapOutputTrackerMaster，两者都是在实例化SparkEnv的时候创建的，每个在Executor上完成的Shuffle Task的结果都会注册到driver端的mapOutputTrackerMaster中，即driver端的mapOutputTrackerMaster的mapStatuses保存这所有元数据信息，所以当一个Executor上的任务需要获取一个shuffle的输出时，会先在自己的mapStatuses中查找，找不到再和mapOutputTrackerMaster通信获取元数据。
mapOutputTrackerMaster收到消息后的处理逻辑：
调用了tracker的post方法：
将该Message加入了mapOutputRequests中，mapOutputRequests是一个链式阻塞队列，在mapOutputTrackerMaster初始化的时候专门启动了一个线程池来执行这些请求：
看看线程处理类MessageLoop的run方法是怎么定义的：
通过shuffleId获取对应序列化后的元数据信息并返回，具体看看的实现：
大体思路是先从缓存中获取元数据（MapStatuses），获取到直接返回，若没有则从mapStatuses获取，获取到后将其序列化后返回，随后返回给mapOutputTrackerWorker（刚才与之通信的节点），mapOutputTracker收到回复后又将元数据序列化并加入当前Executor的mapStatuses中。
再回到getMapSizesByExecutorId方法中，getStatuses得到shuffleID对应的所有的元数据信息后，通过convertMapStatuses方法将获得的元数据信息转化成形如Seq[(BlockManagerId, Seq[(BlockId, Long)])]格式的位置信息，用来读取指定的分区的数据：
这里的参数statuses:Array[MapStatus]是前面获取的上游stage所有的shuffle Write 文件的元数据，并且是按map端的partitionId排序的，通过zipWithIndex将元素和这个元素在数组中的ID（索引号）组合成键/值对，这里的索引号即是map端的partitionId，再根据shuffleId、mapPartitionId、recePartitionId来构建ShuffleBlockId（在map端的ShuffleBlockId构建中的recePartitionId始终是0，因为一个ShuffleMapTask就一个Block，而这里加入的真正的recePartitionId在后面通过index文件获取对应rece端partition偏移量的时候需要用到），并估算得到对应数据的大小，因为后面获取远程数据的时候需要限制大小，最后返回位置信息。
至此mapOutputTracker.getMapSizesByExecutorId(handle.shuffleId, startPartition, endPartition)方法完成，返回了指定分区对应的元数据MapStatus信息。
在初始化对象ShuffleBlockFetcherIterator的时候调用了其初始化方法initialize()：
先看是怎么区分local blocks和remote blocks的：
区分完local remote blocks后加入到了队列fetchRequests中，并调用fetchUpToMaxBytes()来获取远程数据：
从fetchRequests中取出FetchRequest，并调用了sendRequest方法：
通过shuffleClient的fetchBlocks方法来获取对应远程节点上的数据，默认是通过NettyBlockTransferService的fetchBlocks方法实现的，不管是成功还是失败都将构建SuccessFetchResult & FailureFetchResult 结果放入results中。
获取完远程的数据接着通过fetchLocalBlocks()方法来获取本地的blocks信息：
迭代需要获取的block，直接从blockManager中获取数据，并通过结果数据构建SuccessFetchResult或者FailureFetchResult放入results中，看看在blockManager.getBlockData(blockId)的实现：
再看看getBlockData方法：
根据shuffleId和mapId获取index文件，并创建一个读文件的文件流，根据block的receId（上面获取对应partition元数据的时候提到过）跳过对应的Block的数据区，先后获取开始和结束的offset，然后在数据文件中读取数据。
得到所有数据结果result后，再回到read()方法中：
这里的ShuffleBlockFetcherIterator继承了Iterator，results可以被迭代，在其next()方法中将FetchResult以（blockId，inputStream）的形式返回：
在read()方法的后半部分会进行聚合和排序，和Shuffle Write部分很类似，这里大致描述一下。
在需要聚合的前提下，有map端聚合的时候执行combineCombinersByKey，没有则执行combineValuesByKey，但最终都调用了ExternalAppendOnlyMap的insertAll(iter)方法：
在里面的迭代最终都会调用上面提到的ShuffleBlockFetcherIterator的next方法来获取数据。
每次update&insert也会估算currentMap的大小，并判断是否需要溢写到磁盘文件，若需要则将map中的数据根据定义的keyComparator对key进行排序后返回一个迭代器，然后写到一个临时的磁盘文件，然后新建一个map来放新的数据。
执行完combiners[ExternalAppendOnlyMap]的insertAll后，调用其iterator来返回一个代表一个完整partition数据（内存及spillFile）的迭代器：
跟进ExternalIterator类的实例化：
将currentMap中的数据经过排序后和spillFile数据的iterator组合在一起得到inputStreams ，迭代这个inputStreams ，将所有数据都保存在mergeHeadp中，在ExternalIterator方法的next()方法中将被访问到。
最后若需要对数据进行全局的排序，则通过只有排序参数的ExternalSorter的insertAll方法来进行排序，和Shuffle Write一样的这里就不细讲了。
最终返回一个指定partition所有数据的一个迭代器。
❼ spark进入txt文件的命令
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"
❽ Spark 怎么读文件名
Apache Spark 本身
1.MLlib
AMPLab
Spark最初诞生于伯克利 AMPLab实验室，如今依然还是AMPLab所致力的项目，尽管这些不处于Apache Spark Foundation中，但是依然在你日常的github项目中享有相当的地位。
ML Base
Spark本身的MLLib位于三层ML Base中的最底层，MLI位于中间层，ML Optimizer则处于最为抽象的顶层。
2.MLI
3.ML Optimizer (又称 Ghostface)
Ghostware这个项目在2014年就开始进行了，不过从未对外公布。在这39个机器学习库中，这是唯一一个雾件，之所以能囊括在这列表中，全凭着AMPLab与ML Base的地位支撑。
ML Base之外
4.Splash
这是近期2015年6月的一个项目，在运行随机梯度下降（SGD）时这套随机学习算法声称在性能上比Spark MLib中快了25%-75%。这是AMPLab实验室的sp标记项目，因此值得我们去阅读。
5.Keystone ML
KML将端到端的机器学习管道引进到了Spark中，但在近期Spark版本中管道已经趋于成熟。同样也承诺具有一些计算机视觉能力，我曾经在博客中也提到过这也存在一些局限。
6.Velox
作为一个服务器专门负责管理大量机器学习模型的收集。
7.CoCoA
通过优化通信模式与shuffles来实现更快的机器学习，详情可见这篇论文的描述《高效通信分布式双坐标上升》。
框架
GPU-based
8.DeepLearning4j
我曾经的一则博客有进行说明 《DeepLearning4J 增加了Spark gpu的支持》。
9.Elephas
全新的概念，这也是我写这篇博客的初衷。它提供了一个接口给Keras。
Non-GPU-based
10.DistML
模式并行下而并非数据并行的参数服务器（正如 Spark MLib）。
11.Aerosolve
来自Airbnb，用于他们自动化定价。
12. Zen
逻辑斯谛回归、隐含狄利克雷分布（LDA）、因子分解机、神经网络、受限玻尔兹曼机。
13.Distributed Data Frame
与Spark DataFrame类似，但是引擎是不可知的（例如在未来它将运行在引擎上而不是Spark）。其中包括了交叉验证和外部机器学习库的接口。
其他机器学习系统的接口
14. spark-corenlp
封装了斯坦福CoreNLP。
15. Sparkit-learn
给Python Scikit-learn的接口。
16. Sparkling Water
给 的接口。
17. hivemall-spark
封装了Hivemall,，在Hive中的机器学习。
18. spark-pmml-exporter-validator
可导出预测模型标记语言（PMML），一种用于传递机器学习模型的行业标准的XML格式。
附加组件：增强MLlib中现有的算法。
19. MLlib-dropout
为Spark MLLib 增加dropout能力。基于以下这篇论文进行的实现，《Dropout：一个简单的方法来防止神经网络中的过拟合》。
20.generalized-kmeans-clustering
为K-Means算法增加任意距离函数。
21. spark-ml-streaming
可视化的流式机器学习算法内置于Spark MLlib。
算法
监督学习
22. spark-libFM
因子分解机。
23. ScalaNetwork
递归神经网络（RNNs）。
24. dissolve-struct
基于上文中提到的高性能Spark通信框架CoCoA下的支持向量机（SVM）。
25. Sparkling Ferns
基于以下这篇论文进行的实现，《通过使用随机森林与随机蕨算法的图像分类技术》。
26. streaming-matrix-factorization
矩阵分解推荐系统。
❾ spark与hdfs怎么加载本地文件
利用Spark构建一个树索引结构。具体内容如下：
（1）数据形式是20-dimension 的double向量
（2）树的结构是：
1>内部结点都保存一个数据对象和指向子树的指针，每个子树中数据对象与父节点中保留的对象的距离由远到近不同
2>叶子结点保存数据对象的列表
（3）想通过Spark加载数据，将整个数据源文件抽象为一个RDD， RDD分区存储文件的不同块，然后写一个索引构建算子作用到每个分区上，然后构建的索引写到文件中（索引文件的个数与RDD分区的个数相同）
（4）查找时查找算子作用到每个索引文件
现在有一个问题就是在构建索引时一个分区中的源数据是满的，构建的索引数据大于一个分区的空间（因为空间占用方面，索引数据肯定大于源数据），索引数据会写到不同的分区中，查找过程中，查找算子是以分区为单位进行查找的，但是单个分区数据不是一个完整的索引
默认是从hdfs读取文件，也可以指定sc.textFile("路径").在路径前面加上hdfs://表示从hdfs文件系统上读
本地文件读取 sc.textFile("路径").在路径前面加上file:// 表示从本地文件系统读，如file:///home/user/spark/README.md
❿ spark怎样读取本地文件
默认是从hdfs读取抄文件，也可以指定sc.textFile("路径").在路径前面加上hdfs://表示从hdfs文件系统上读
本地文件读取 sc.textFile("路径").在路径前面加上file:// 表示从本地文件系统读，如file:///home/user/spark/README.md