将org.apache.spark.rdd.RDD[String]转换为并行化集合
将org.apache.spark.rdd.RDD[String]转换为并行化集合，可以使用Spark的collect()方法。
RDD（弹性分布式数据集）是Spark中的核心数据结构，代表了一个分布式的不可变数据集。RDD可以通过一系列的转换操作进行处理，例如过滤、映射、聚合等。而collect()方法是RDD的一个动作操作，用于将RDD中的数据收集到驱动程序中，并以数组的形式返回。
在将RDD[String]转换为并行化集合之前，需要先创建一个SparkContext对象，并通过该对象创建一个RDD。假设已经创建了一个名为rdd的RDD对象，可以使用以下代码将其转换为并行化集合：
val sparkConf = new SparkConf().setAppName("Example").setMaster("local")
val sc = new SparkContext(sparkConf)
val rdd: org.apache.spark.rdd.RDD[String] = ???
val collection: Array[String] = rdd.collect()
// 打印并行化集合中的元素
collection.foreach(println)
// 关闭SparkContext
sc.stop()在上述代码中，首先创建了一个SparkConf对象，用于配置Spark应用程序的相关参数，例如应用程序名称和运行模式。然后，通过SparkConf对象创建了一个SparkContext对象，用于与Spark集群进行通信。
接下来，假设已经存在一个RDD对象rdd，可以使用rdd.collect()方法将其转换为并行化集合。collect()方法会将RDD中的数据收集到驱动程序中，并以数组的形式返回。最后，可以通过遍历并行化集合中的元素，对其进行进一步处理。
需要注意的是，在使用完SparkContext对象后，需要调用stop()方法来关闭SparkContext，释放资源。
关于Spark的更多信息和相关产品介绍，可以参考腾讯云的Spark产品页面：Spark - 腾讯云
相关·内容
要将函数映射到RDD的所有元素，需要首先使用collect方法将RDD转换为Array类型：
scala> val x = sc.parallelize(List(List("a"), List("b"), List("c", "d")))
x: org.apache.spark.rdd.RDD[List[String]] = ParallelCollectionRDD[1] at parallelize at <console>:12
scala> x.collect()
res0: Array[Lis
浏览 3提问于2014-05-21得票数 2
回答已采纳
我正在使用Spark2.1.0和Scala2.10.6
当我尝试这样做的时候：
val x = (avroRow1).join(flattened)
我知道错误：
value join is not a member of org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]
我为什么要收到这条消息？我有下列进口报表：
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
impor
浏览 0提问于2017-07-10得票数 0
1回答
我正在尝试下面的例子
val lista = List(("a", 3), ("a", 1), ("b", 7), ("a", 5))
val rdd = sc.parallelize(lista)
然后在shell中我得到以下信息
rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[40] at parallelize at <console>:30
但出于某种原因，我仍然没有意识到我能执行这句话
val resAgg = rdd.
浏览 4提问于2016-07-20得票数 0
我的HDFS中有一个csv文件，其中包含一系列产品，例如：
[56]
[85,66,73]
[57]
[8,16]
[25,96,22,17]
[83,61]
我正在尝试在我的代码中应用关联规则算法。为此，我需要运行以下命令：
scala> val data = sc.textFile("/user/cloudera/data")
data: org.apache.spark.rdd.RDD[String] = /user/cloudera/data MapPartitionsRDD[294] at textFile at <console>:38
scal
浏览 23提问于2016-09-26得票数 1
回答已采纳
我已经建立了火花核心项目从。我调用了一个测试类：CacheManagerSuite，它通过了。
如何在源上运行一些Spark转换/操作？为了在下面运行，我需要在Spark项目源代码中调用哪些类/对象？
scala> val x = sc.parallelize(List(List("a"), List("b"), List("c", "d")))
x: org.apache.spark.rdd.RDD[List[String]] = ParallelCollectionRDD[1] at parallelize at <
浏览 3提问于2014-05-26得票数 2
回答已采纳
2回答
要将文件读入内存，我使用以下命令：
val lines = sc.textFile("myLogFile*")
它的类型是：
org.apache.spark.rdd.RDD[String] = MappedRDD[1] at textFile at <console>:12
阅读Scala文档：“并行集合是通过在现有的Scala集合(序列对象)上调用SparkContext的parallelize方法创建的。”
这似乎不适用于RDD？RDD上可以进行并行化处理吗？我是否需要将RDD转换为Seq对象？
浏览 0提问于2014-04-26得票数 6
回答已采纳
我试图理解map和flatMap是如何工作的，但被下面的代码卡住了。flatMap()函数返回一个RDDChar，但我期望返回的是RDDString。有人能解释一下为什么它会产生RDDChar吗？
scala> val inputRDD = sc.parallelize(Array(Array("This is Spark"), Array("It is a processing language"),Array("Very fast"),Array("Memory operations")))
scala> val
浏览 17提问于2017-06-25得票数 1
回答已采纳
我通过指定分区数从文本文件创建RDD。但是它给出了与指定分区不同的分区数。
scala> val people = sc.textFile("file:///home/pvikash/data/test.txt", 0)
people: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[72] at textFile at <console>:27
scala> people.getNumPartitions
res47: Int = 1
scala> val people = sc.
浏览 0提问于2018-01-18得票数 0
回答已采纳
根据星火源代码注释。
SparkContext.scala有
/** Distribute a local Scala collection to form an RDD.
*
* @note Parallelize acts lazily. If `seq` is a mutable collection and is altered after the call
* to parallelize and before the first action on the RDD, the resultant RDD will reflect the
* modifie
浏览 1提问于2016-11-28得票数 1
回答已采纳
我在scala中有以下实用程序函数：
object MyUtiltity {
def processData(data1: org.apache.spark.rdd.RDD[String], data2: org.apache.spark.rdd.RDD[String],
data3: org.apache.spark.rdd.RDD[String]) = {
function1(data1, data3)
function2(data2, data3)
}
private def function1 {...}
private def fun
浏览 0提问于2015-06-05得票数 0
回答已采纳
我有一个字符串的ArrayBuffer，它包含了我想要创建的图的所有顶点的标签。我需要创建一个RDD对象[(VertexId, String)]，它将成为我未来图形的节点，其中每个节点的VertexId = ArrayBuffer中节点标签的索引。我只找到了有关使用SparkContext.textFile(String fname)创建RDD的信息，但没有发现任何关于如何从数据结构创建RDD的信息。
有办法做到这一点吗?还是我总是必须从文件中创建RDD？
浏览 3提问于2015-10-18得票数 1
回答已采纳
我以DataFrame array<string>的身份处理事务
transactions: org.apache.spark.sql.DataFrame = [collect_set(b): array<string>]
我想将其更改为RDD[Array[string]]，但是当我将其更改为RDD时，它被更改为org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]
val sam: RDD[Array[String]] = transactions.rdd
<console>:42: error: type m
浏览 14提问于2017-01-11得票数 1
sessionIdList的类型为：
scala> sessionIdList
res19: org.apache.spark.rdd.RDD[String] = MappedRDD[17] at distinct at <console>:30
当我尝试运行下面的代码时：
val x = sc.parallelize(List(1,2,3))
val cartesianComp = x.cartesian(x).map(x => (x))
val kDistanceNeighbourhood = sessionIdList.map(s => {
ca
浏览 0提问于2014-05-22得票数 14
回答已采纳
1回答
我试图编写一个函数来对RDD[SeqString]对象进行操作，例如：
def foo(rdd: RDD[Seq[String]]) = { println("hi") }
不能对RDD[ArrayString]类型的对象调用此函数：
val testRdd : RDD[Array[String]] = sc.textFile("somefile").map(_.split("\\|", -1))
foo(testRdd)
->
error: type mismatch;
found : org.apache.spark.rdd.RD
浏览 2提问于2014-05-22得票数 7
回答已采纳
2回答
我面对的是flatMap的错误，而不是map的错误。flatMap转换不支持布尔操作吗？
scala> val array = Array("age","astro")
array: Array[String] = Array(age, astro)
scala> val baseRdd = sc.parallelize(array)
baseRdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at <console>:29
s
浏览 7提问于2016-05-29得票数 0
回答已采纳
我需要parallelize我的数据集，但是，当我试图使用重新分区的数据实例化一个RowMatrix时，会发生type mismatch错误。
下面是读取和重新分区数据的代码：
val data = sc.textFile("data.txt.gz").flatMap(r => r.split(' ') match {
case Array(doc, word, count) => Some((doc.toInt, (word.toInt - 1, count.toDouble)))
case _ => None
}).groupByK
浏览 3提问于2017-04-20得票数 0
回答已采纳
1回答
我对join或cogroup方法在RDD上有问题。详细地说，我必须加入两个RDD，其中一个是泛型类型的RDD，与通配符一起使用。
val indexedMeasures = measures.map(m => (m.id(), m)) // RDD[(String, Measure[_]]
val indexedRegistry = registry.map(r => (r.id, r)) // RDD[(String, Registry)]
indexedRegistry.cogroup(indexedMeasures)
最后一条语句给出了编译时错误，如下所示：
no typ
浏览 2提问于2016-01-20得票数 2
回答已采纳
我正在Google上运行一个Spark集群，在尝试使用GZipped读取sparkContext.textFile(...)文件时遇到了一些问题。
我正在运行的代码是：
object SparkFtpTest extends App {
val file = "ftp://username:password@host:21/filename.txt.gz"
val lines = sc.textFile(file)
lines.saveAsTextFile("gs://my-bucket-storage/tmp123")
}
我得到的错误是：
Exc
浏览 5提问于2016-12-06得票数 0
回答已采纳
我有一个RDD (字符串数组) org.apache.spark.rdd.RDD[String] = MappedRDD[18]，并将其转换为具有唯一Ids的映射。我做了'val vertexMAp = vertices.zipWithUniqueId‘，但这给了我另一个'org.apache.spark.rdd.RDD[(String, Long)]'类型的RDD，但我想要一个'Map[String, Long]’。如何转换我的'org.apache.spark.rdd.RDD[(String, Long)] to Map[String, Long]‘
浏览 4提问于2014-10-14得票数 5
回答已采纳
我正在尝试在spark环境中使用scala读取avro数据。我的数据没有被分发，并且在运行时只会被分配到2个节点上。我们有20+nodes。下面是我的代码片段
@serializable case class My_Class (val My_ID : String )
val filePath = "hdfs://path";
val avroRDD = sc.hadoopFile[AvroWrapper[GenericRecord], NullWritable, AvroInputFormat[GenericRecord]](filePath)
val rddprsi
浏览 2提问于2015-10-20得票数 0
我有以下建议：
res38: org.apache.spark.rdd.RDD[(String, Iterable[String])] = ShuffledRDD[5] at groupBy at <console>:14
现在，我想得到这个元组的第二个元素，并将它们放在一个数组中：
所以我试着：
scala> val items = data.map(x => x._2.toArray)
items: org.apache.spark.rdd.RDD[Array[String]] = MappedRDD[17] at map at <console>:16
浏览 4提问于2015-03-10得票数 1
回答已采纳
我有这个数据集(我放了几行)：
11.97,1355,401
3.49,25579,12908
9.29,129186,10882
28.73,10153,22356
3.69,22872,9798
13.49,160371,2911
24.36,106764,867
3.99,163670,16397
19.64,132547,401
我试图用K-均值将所有这些行分配给4个集群。为此，我使用了我在本文中看到的代码：
val data = sc.textFile("/user/cloudera/TESTE1")
val idPointRDD = data.map(s =>
浏览 0提问于2016-10-30得票数 0
回答已采纳
使用新的Stream进行简单迭代有什么好处吗？
没有Stream：
for (Map.Entry<String, String> entry : map.entrySet()) {
doSomething(entry);
}
使用Stream：
map.entrySet().stream().forEach((entry) -> {
doSomething(entry);
});
代码的长度和可读性大致相同。是否有任何重要的差异(例如在性能方面)？
浏览 0提问于2014-04-24得票数 5
回答已采纳
我正在尝试使用scala并行性来加速执行时间。
因此，为了将java ArrayList转换为不可变的java use，我使用：
var imList = scala.collection.JavaConversions.asScalaBuffer(normalQLFolderList)
然后，为了在迭代时利用多个内核，我使用：
for (i <- imList .par) {
}
我是否以正确的方式利用了scala并行性？在这种情况下，遍历列表。在asScalaBuffer上有很大的性能冲击吗？
浏览 2提问于2013-02-22得票数 7
回答已采纳
我有大量的URL列表(随着时间的推移而扩展)，我需要检查它们的类型。这就是我目前的代码：
private string[] MIME = new string[] {
"audio/ogg - ogg",
"video/ogg - ogg",
"application/f4v - mp4",
"application/octet-stream - mp3",
"audio/aac - mp3",
"aud
浏览 3提问于2011-09-09得票数 0
回答已采纳
1回答
查找屏幕上键盘scala的击键
、、、、
我正试着用Scala来解决最近的一个面试问题。
您有一个屏幕上的键盘，它是一个6行的网格，每列5列。使用从A到Z的字母表，空格首先排列在网格行中。
您可以在屏幕键盘上使用这个输入单词。使用您的电视遥控器按左、右、上、下或确定键键入每个字符。
问题:给定一个输入字符串，找到需要按在遥控器上键入输入的键击序列。
代码实现可以在
我试着用三种不同的方法来解决这个问题。
简单的forldLeft。
def keystrokesByFL(输入: String，startChar: Char = 'A')：String ={ val = (""，startChar)
浏览 0提问于2017-05-11得票数 0
回答已采纳
2回答
我已经得到了一些优化的代码。其中一个位包含一些代码，这些代码使用带有元素的集合，对于集合中的所有元素，将它们与所有其他元素进行比较。比较是不对称的，所以没有捷径。代码如下：
for(String string : initialSet)
{
Set<String> copiedSet = new HashSet<>(initialSet);
copiedSet.remove(string);
for(String innerString : copiedSet)
{
/**
* Magic, uni
浏览 2提问于2013-12-19得票数 2
回答已采纳
下面的linq代码可以正确地为最上面的特殊节点的子节点提供记录组。这是如何使用lambda的？
最上面的查询从数据库中获取view_consulting记录，并根据speciality_name正确地将它们分组：
// get the records from the database.
view_consulting[] v = MyNetwork.Medical.Client.GetConsultingStaff();
// All records will first be grouped by the speciality name.
浏览 1提问于2018-08-20得票数 1
回答已采纳
我正在使用Spark，并创建了两个idential长度数组，一个是tweet的时间，另一个是tweet的文本。我希望将它们合并成一个数据结构(也许是一个元组？)我可以过滤的时间和文字的推特，但我正在挣扎后，结合起来，如何执行。
scala> val split_time = split_date.map(line => line.split(":")).map(word =>
(word(0)))
split_time: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[28] at map
at <
浏览 0提问于2017-12-08得票数 1
回答已采纳
2回答
我有大约100个硒测试要运行，但其中两个不能在并行中运行
是否可以仅对这两个测试禁用并行性，使用xUnit?
(这两个测试不能并行，因为它们需要模拟键盘单击->，这样我就会失去使用并行执行的输入焦点)
我正在寻找的最佳场景：
向两个测试添加一些属性，这些测试将禁用它们的并行性。然后，在测试中，执行98个测试将运行16个线程，剩下的两个测试将在最后使用1个线程执行。
我知道其中一个解决方案可以是这样的：
为测试添加“并行”和“NonParallel”类别
使用xunit.console，只运行参数为maxthread=16的“并行”类别。
之后，使用参数NonParallel运
浏览 0提问于2018-02-07得票数 12
2回答
也许我遗漏了一些东西，但我希望数据会根据键进行排序。
scala> val x=sc.parallelize(Array( "cat", "ant", "1"))
x: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[160] at parallelize at <console>:22
scala> val xxx=x.map(v=> (v,v.length))
xxx: org.apache.spark.rdd.RDD[(String, Int)
浏览 2提问于2016-01-03得票数 3
回答已采纳
我试图用逗号分隔符连接Scala中的XML属性。
scala> val fileRead = sc.textFile("source_file")
fileRead: org.apache.spark.rdd.RDD[String] = source_file MapPartitionsRDD[8] at textFile at <console>:21
scala> val strLines = fileRead.map(x => x.toString)
strLines: org.apache.spark.rdd.RDD[String] =
浏览 1提问于2015-12-09得票数 0
回答已采纳
..。
val cols: Seq[String] = Seq("item", "SR", "RP")
val vecToSeq = udf((v:org.apache.spark.ml.linalg.Vector) => v.toArray)
val exprs = cols.zipWithIndex.map{ case(c,i) => $"_tmp".getItem(i).alias(c)}
val DoubleDF = result5.select(vecToSeq($"vectorCol").
浏览 0提问于2017-02-28得票数 0
1回答
我只在一个具有Spark的节点上运行Parallelism = 1，以便将其性能与单线程应用程序进行比较。我想知道Spark是否仍然在使用Shuffle，尽管它不是并行运行的。因此，如果执行以下命令：
val counts = text_file.flatMap(line => line.split(" "))
.map(word => (word, 1))
.reduceByKey(_+_)
我从获得以下输出：
counts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[10]
浏览 1提问于2015-12-13得票数 3
回答已采纳
2回答
我在Scala中使用Spark。我想创建一个图形并动态更新该图形。
我用下面的代码做到了这一点：
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
object firstgraph {
def addVertex(
sc: SparkContext,
vertexRDD: R
浏览 1提问于2015-12-08得票数 1
Pg_dump有两种主要格式:自定义vs。对于自定义，默认情况下它是压缩的，如果尝试打开它，则无法读取。但是，与SQL格式相比，将其转储到这种格式会更快吗？还是速度是一样的？
浏览 0提问于2013-08-30得票数 5
集合框架的大部分实现复杂性源于这样一个事实，即Scala可以-不像C#的LINQ或其他集合框架-返回高阶函数的“最佳”集合类型：
val numbers = List(1,2,3,4,5)
numbers map (2*) // returns a List[Int] = List(2, 4, 6, 8)
val doubles = Array(1.0, 2.0, 3.0)
doubles filter (_ < 3) // returns Array[Double] = Array(1.0, 2.0)
为什么这一原则不适用于seq、par、view、force等方法
numbers.
浏览 8提问于2011-06-18得票数 8
回答已采纳
1回答
我有一个密钥对映射(String，Set(Integer))。其中每个条目表示一个元素的出现。现在我要做的是从那个键中得到一组元素，并找到所有这些同时发生的地方。集合将逐渐增加。
我被困在如何形成一个方法的字符串集和映射字符串，集-整数作为输入，将返回一组整数作为输出，将包含所有整数集合在映射对应于一组字符串的键的交集。
这是我的地图
Map<String,Set<Integer>> transactions = new HashMap<String,Set<Integer>>();
这是一组字符串
Set<String> check
浏览 3提问于2014-06-16得票数 0
我用他们的路由创建了一个自定义实体"Faq“。当我在我的index.html.twig中转储“常见问题”时，Faq#translations#collection是空的。
我需要集合中的一个“问题”和一个“响应”。在我的数据库中，所有字段都存在于FaqTranslation中。
我可能错过了什么？
对不起，我的英语很差。
谢谢。
结果转储(Faqs)
Faq {#3359 ▼
-id: 2
#translations: PersistentCollection {#3369 ▼
-snapshot: []
-owner: Faq {#3359}
-association: array:
浏览 3提问于2017-05-09得票数 0
如果我不包含“可序列化的实现”，会发生什么？
public class Student implements Serializable {
private String studentNumber;
private String firstName;
private String lastName;
private ArrayList<Exam> exams;
}
浏览 0提问于2015-11-12得票数 8
我有一个ASP MVC控制器联系了一堆http-站点在一个for-循环.比如：
foreach(string provider in providers){
// get data from asomehttp URL
}
这需要大约4秒的时间.
我试过：
Parallel.ForEach(providers, (provider) => {
// get data from some http URL
});
而且我没有看到任何表现上的提高！
为什么会这样呢？
浏览 1提问于2016-12-22得票数 0
我正在尝试使用字符串的rdd作为字典和来自包的类org.apache.spark.mllib.random来创建一个字符串采样器。
import org.apache.spark.mllib.random.RandomDataGenerator
import org.apache.spark.rdd.RDD
import scala.util.Random
class StringSampler(var dic: RDD[String], var seed: Long = System.nanoTime) extends RandomDataGenerator[String] {
r
浏览 4提问于2017-07-28得票数 1
我在堆内存中有键列表，值对，如List((A,1)，(B,2)，(C,3))。如何并行处理这个列表以创建一个JavaPairRDD？在scala中: val对=sc.parallelize(List(A，1)，(B,2)，(C,3))。同样，java有任何方法吗？
浏览 9提问于2016-04-29得票数 0
回答已采纳
我正在和Spark合作。我需要附加/连接两个类型为Set的RDD。
scala> var ek: RDD[Set[Int]] = sc.parallelize(Seq(Set(7)))
ek: org.apache.spark.rdd.RDD[Set[Int]] = ParallelCollectionRDD[31] at parallelize at <console>:32
scala> val vi: RDD[Set[Int]] = sc.parallelize(Seq(Set(3,5)))
vi: org.apache.spark.rdd.RDD[Set[In
浏览 2提问于2017-10-17得票数 1
回答已采纳
我是Java和Spark的新手，我发现了一个令人印象深刻的库，它为Spark提供了，它允许我们使用C#与SparkSQL一起工作。
我在一个具有ODBC和OPC接口的自定义数据存储中有一些大量的过程数据。我们希望将此数据公开给Apache Spark，以便我们可以使用Apache Zeppelin等工具对此数据运行分析查询
因为我的自定义存储上没有jdbc接口，所以我正在考虑创建c#代码，以便使用可用的ODBC接口从自定义数据存储中提取数据，并使用historyDataFrame.RegisterTempTable("mydata");将其提供给spark。
我可以创建一个示例
浏览 5提问于2016-01-05得票数 2
我做了以下的火花弹练习：
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 2.1.0
/_/
Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_121)
Type in expressions to have them evaluated.
Type :hel
浏览 4提问于2017-08-01得票数 1
回答已采纳
我最近了解到了Java语言中用于并发的各种库，比如和来自的可爱的非阻塞库
我对Scala了解不多，但我听说了最近关于并行集合库的一些好消息。
我想知道这个库与基于Java的库相比有哪些主要优势？
浏览 1提问于2011-06-02得票数 12
回答已采纳
我使用一条命令从解决方案级别运行所有测试项目：dotnet test如何使所有测试项目(程序集)并行运行？
在Visual Studio中，有一个简单的按钮“并行运行测试”，它工作得很好，但我需要使用dotnet核心测试命令来执行CI。
浏览 6提问于2018-12-28得票数 13
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
