将 JavaPairRDD 转换为 JavaRDD
我正在使用 ElasticSearch-Hadoop 库从 ElsticSearch 获取数据. JavaPairRDD>esRDD = JavaEsSpark.esRDD(sc); 现在我有了 JavaPairRDD.我想在这个 RDD 上使用来自 MLLib 的随机森林.所以我将它转换为 JavaPairRDD.toRDD(esRDD) 这会给我 RDD.使用 RDD 我再次转换为 Jav
..
rdd相关内容
如何使用RDD分组和聚合多个字段?
我是 Apache Spark 和 Scala 的新手，目前正在学习这个框架和大数据编程语言.我有一个示例文件，我试图找出另一个字段的给定字段总数及其计数和来自另一个字段的值列表.我自己尝试过，似乎我没有在 spark rdd(作为开始)中以更好的方法编写. 请找到以下示例数据(Customerid: Int, Orderid: Int, Amount: Float): 44,8602,3
..
Spark RDD:如何最有效地计算统计数据?
假设存在类似于以下元组的 RDD: (key1, 1)(key3, 9)(key2, 3)(key1, 4)(key1, 5)(key3, 2)(key2, 7)... 计算与每个键对应的统计信息的最有效(并且理想情况下是分布式)方法是什么?(目前，我希望计算标准偏差/方差，特别是.)据我所知，我的选择是: 使用 MLLib 中的colStats 函数: 这种方法的优点是很容易适应以后使
..
将 RDD[org.apache.spark.sql.Row] 转换为 RDD[org.apache.spark.mllib.linalg.Vector]
我对 Spark 和 Scala 比较陌生. 我从以下数据框开始(由密集的双精度向量组成的单列): scala>val scaledDataOnly_pruned = scaledDataOnly.select("features")scaledDataOnly_pruned: org.apache.spark.sql.DataFrame = [features: vector]标度>sc
..
Spark Mlib FPGrowth 作业因内存错误而失败
我有一个相当简单的用例，但结果集可能非常大.我的代码执行以下操作(在 pyspark shell 上): from pyspark.mllib.fpm import FPGrowthdata = sc.textFile("/Users/me/associationtestproject/data/sourcedata.txt")交易 = data.map(lambda line: line.st
..
RDD 到 LabeledPoint 的转换
如果我有一个大约有 500 列和 2 亿行的 RDD，并且 RDD.columns.indexOf("target", 0) 显示 Int = 77我的目标因变量在第 77 列.但我对如何选择所需的(部分)列作为特征没有足够的知识(比如我想要从 23 到 59、111 到 357、399 到 489 的列).我想知道我是否可以申请这样的: val data = rdd.map(col => ne
..
Spark ALS predictAll 返回空
我有以下 Python 测试代码(ALS.train 的参数在别处定义): r1 = (2, 1)r2 = (3, 1)测试 = sc.parallelize([r1, r2])模型 = ALS.train(评级，排名，numIter，lmbda)预测 = model.predictAll(test)打印 test.take(1)打印predictions.count()打印预测 哪个有效，
..
Apache Spark 中的矩阵乘法
我正在尝试使用 Apache Spark 和 Java 执行矩阵乘法. 我有两个主要问题: 如何在 Apache Spark 中创建可以表示矩阵的 RDD? 如何将两个这样的 RDD 相乘? 解决方案 一切都取决于输入的数据和维度 但一般来说你想要的不是 RDD 而是一种分布式数据结构org.apache.spark.mllib.linalg.distributed.目前它提
..
使用 Java 将 spark RDD 保存到本地文件系统
我有一个使用 Spark 生成的 RDD.现在，如果我将此 RDD 写入 csv 文件，则会提供一些方法，例如“saveAsTextFile()"，该方法将 csv 文件输出到 HDFS. 我想将文件写入我的本地文件系统，以便我的 SSIS 进程可以从系统中选取文件并将它们加载到数据库中. 我目前无法使用 sqoop. 除了编写 shell 脚本之外，在 Java 中是否有可能做
..
将压缩在 tar.gz 存档中的多个文件读入 Spark
我正在尝试从压缩成 tar 的几个 json 文件创建一个 Spark RDD.例如，我有 3 个文件 file1.json文件2.json文件3.json 这些都包含在archive.tar.gz中. 我想从 json 文件创建一个数据框.问题是 Spark 没有正确读取 json 文件.使用 sqlContext.read.json("archive.tar.gz") 或 sc.te
..
如何在 Spark 中转置 RDD
我有一个这样的 RDD: 1 2 34 5 67 8 9 这是一个矩阵.现在我想像这样转置RDD: 1 4 72 5 83 6 9 我该怎么做? 解决方案 假设你有一个 N×M 矩阵. 如果 N 和 M 都小到可以在内存中保存 N×M 个项目，那么使用 RDD 就没有多大意义了.但是转置很容易: val rdd = sc.parallelize(Seq(Seq(1, 2,
..
修改 Spark RDD foreach 中的集合
我正在尝试在迭代 RDD 的元素时向地图添加元素.我没有收到任何错误，但没有发生修改. 直接添加或迭代其他集合一切正常: scala>val myMap = new collection.mutable.HashMap[String,String]myMap: scala.collection.mutable.HashMap[String,String] = Map()标度>myMap("
..
为什么 Spark RDD 分区对 HDFS 有 2GB 的限制?
我在使用 mllib RandomForest 训练数据时遇到错误.由于我的数据集很大，而默认分区相对较小.所以抛出异常表明“大小超过Integer.MAX_VALUE"，原始堆栈跟踪如下， 15/04/16 14:13:03 WARN scheduler.TaskSetManager: Lost task 19.0 in阶段 6.0(TID 120、10.215.149.47):java.
..
如何在 Spark RDD (Java) 中按索引获取元素
我知道 rdd.firstwfirst() 方法，它为我提供 RDD 中的第一个元素. 还有一个方法 rdd.take(num) 它给了我第一个“num"元素. 但是不是有可能通过索引获取元素吗? 谢谢.e 解决方案 这应该可以通过首先索引 RDD 来实现.转换 zipWithIndex 提供了稳定的索引，按其原始顺序对每个元素进行编号. 给定:rdd = (a,b
..
如何计算合并的最佳分区数?
所以，我明白通常应该在以下情况下使用 coalesce(): 由于 filter 或其他一些可能导致原始数据集(RDD，DF)减少的操作，分区数量减少.coalesce() 对于过滤大型数据集后更有效地运行操作非常有用. 我也知道它比 repartition 更便宜，因为它仅在必要时通过移动数据来减少混洗.我的问题是如何定义 coalesce 需要的参数 (idealPartionNo
..
当文件无法放入 spark 的主内存时，spark 如何读取大文件(PB)
在这些情况下，大文件会发生什么? 1) Spark 从 NameNode 获取数据的位置.根据 NameNode 的信息，Spark 会在同一时间因为数据太长而停止吗? 2) Spark 根据数据节点块大小对数据进行分区，但所有数据无法存储到主内存中.这里我们没有使用 StorageLevel.那么这里会发生什么呢? 3) Spark 对数据进行分区，一些数据会存储在主内存中，一
..
使用 Python 计算 Spark 中 Pairwise (K,V) RDD 中每个 KEY 的平均值
我想将这个特殊的 Apache Spark 与 Python 解决方案分享，因为它的文档很差. 我想通过 KEY 计算 K/V 对(存储在 Pairwise RDD 中)的平均值.示例数据如下所示: >>>rdd1.take(10) # 展示一个小样本.[(u'2013-10-09', 7.60117302052786),(u'2013-10-10', 9.322709163346612)
..
如何控制RDD分区的首选位置?
有没有办法手动设置RDD分区的首选位置?我想确保在某台机器上计算某个分区. 我正在使用数组和“Parallelize"方法从中创建 RDD. 另外我没有使用 HDFS，文件在本地磁盘上.所以我要修改执行节点. 解决方案 有没有办法手动设置RDD分区的preferredLocations? 是的，有，但它是特定于 RDD 的，因此不同类型的 RDD 有不同的方法.
..
为什么 sortBy 转换会触发 Spark 作业?
根据 Spark 文档，只有 RDD 操作可以触发 Spark 作业，并且在对其调用操作时会延迟评估转换. 我看到 sortBy 转换函数被立即应用，它在 SparkUI 中显示为作业触发器.为什么? 解决方案 sortBy 是使用 sortByKey 实现的，它依赖于 RangePartitioner (JVM) 或分区函数 (Python).当您调用 sortBy/sortByK
..
为什么 SparkContext.textFile 的 partition 参数没有生效?
scala>val p=sc.textFile("file:///c:/_home/so-posts.xml", 8)//我有 8 个内核p: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[56] at textFile at :21标度>p.partitions.sizeres33:整数 = 729 我原以为会打印 8 个，但在 Spa
..