从元组RDD中删除空字符串
可以通过以下步骤实现：
- 首先，将元组RDD转换为普通RDD，以便可以对其进行操作。
- 使用过滤函数对RDD进行过滤，以删除空字符串。过滤函数可以使用lambda表达式来定义。
- 最后，将过滤后的RDD转换回元组RDD。
以下是完善且全面的答案：
在云计算中，元组（Tuple）是一种数据类型，用于将多个元素组合成一个集合。元组RDD是分布式计算框架中的一种数据结构，可以包含多个元组，并在分布式环境下进行处理和分析。
删除空字符串是一个常见的数据处理任务，可以使用云计算平台提供的各种工具和技术来实现。在处理元组RDD时，我们可以使用分布式计算框架提供的操作函数来过滤出不包含空字符串的元组。
具体操作如下：
- 首先，我们可以将元组RDD转换为普通RDD，以便可以对其进行操作。在大多数分布式计算框架中，元组RDD可以通过map操作进行转换。
- 接下来，我们可以使用过滤函数对RDD进行过滤，以删除空字符串。过滤函数可以使用lambda表达式来定义，以便对每个元组进行判断和过滤。在lambda表达式中，我们可以使用条件语句来判断字符串是否为空，并将其与空字符串进行比较。
- 最后，我们可以将过滤后的RDD转换回元组RDD，以便进行后续的处理和分析。在大多数分布式计算框架中，元组RDD可以通过reduce操作进行合并和汇总。
对于腾讯云的用户，可以使用腾讯云提供的分布式计算服务，例如腾讯云数据计算服务（Tencent Cloud Data Processing，CDP），来处理元组RDD并删除空字符串。CDP提供了强大的分布式计算引擎和工具，可以帮助用户高效地进行数据处理和分析。
腾讯云数据计算服务产品介绍链接：https://cloud.tencent.com/product/cdp
总结起来，从元组RDD中删除空字符串可以通过将RDD转换为普通RDD、使用过滤函数进行过滤、再将RDD转换回元组RDD来实现。腾讯云的数据计算服务（CDP）是一个推荐的解决方案，可以帮助用户高效地进行此类数据处理任务。
相关·内容
有任何方法可以将一对RDD转换回常规RDD吗？
假设我得到一个本地csv文件，并且首先将它作为一个常规的rdd加载。
rdd = sc.textFile("$path/$csv")
然后创建一对rdd (即键是"，“之前的字符串，而值是"，”之后的字符串)
pairRDD = rdd.map(lambda x : (x.split(",")[0], x.split(",")[1]))
我使用pairRDD ()存储saveAsTextFile()
pairRDD.saveAsTextFile("$savePath"
浏览 1提问于2015-10-06得票数 1
回答已采纳
我正在使用sc.textFile()从Mahout中的Alluxio中获取数据，但它是spark RDD。我的程序进一步使用这个spark RDD作为Mahout DRM，因此我需要将RDD转换为DRM。因此，我当前的代码保持稳定。
浏览 2提问于2017-04-07得票数 3
我一直试图阅读我的数据从一个卡夫卡的主题，并将它写到一个拼花文件。到目前为止，除了foreachRdd函数之外，一切都在工作。当我在dstream上使用映射时，我可以看到数据，但是在使用foreachRdd的下一步，Rdd总是空的，我不知道为什么。
我的环境是Ubuntu运行卡夫卡和火花独立。我用的是火花放电壳。我对python还不熟悉，所以在语法方面我仍然遇到了很多问题，我不知道这是否是我的问题所在。
任何帮助或洞察力都将不胜感激。
下面是我的代码的一个副本，我一直把它粘贴在pyspark外壳中
from pyspark import SparkContext
from pyspark.st
浏览 3提问于2018-01-07得票数 1
回答已采纳
我正在使用GraphX API来构建一个图形，并使用Pregel进行处理。如果从vprog函数返回参数元组，则不会发生错误，但如果使用相同的元组返回新元组，则会得到空点错误。以下是相关代码：
val verticesRDD = cleanDtaDF.select("ChildHash", "DN").rdd.map(row => (row(0).toString.toLong, (row(1).toString.toDouble,row(0).toString.toLong)))
val edgesRDD = (rawDtaDF.select("
浏览 0提问于2018-03-11得票数 0
回答已采纳
我想要做的事情：，我正在使用Spark2.3.0和图形。我有一个类似于菜单结构的简单DF：。当我使用下面的代码创建一个具有三个顶点属性(Int、String、String)的图来创建顶点RDD时，它完美地工作了：
val menuVerticesRDD: RDD[(VertexId, (Int,String,String))] = menuVertexDF
.rdd
.map{row: Row => (row.get(0), row.get(1), row.get(2))}
.map{x => (MurmurHash3.stringHash(x._1.toString).toLon
浏览 1提问于2018-03-27得票数 1
1回答
我正在尝试定义一个函数来获取给定列表的笛卡尔乘积，但是我需要过滤掉包含相同对的元素。
例如:获取rdd的笛卡尔乘积，并过滤出结果((1,0)，(1,0))，((2,0)，(2,0))和((3,0)，(3,0))
rdd = sc.parallelize([(1,0), (2,0), (3,0)])
def get_cart(rdd):
a=sorted(rdd.cartesian(rdd).collect())
aRDD=sc.parallelize(a)
return aRDD
我希望得到输出：
[((1, 0), (2, 0)), ((1, 0),
浏览 0提问于2021-01-24得票数 1
1回答
我只是想学习PySpark，但对下面两个RDDs之间的区别感到困惑，我知道一个是类型集，一个是列表，但都是RDDs。
rdd = sc.parallelize([('a', 1), ('b', 1), ('a', 3)])
type(rdd)
和
rdd = sc.parallelize(['a, 1', 'b, 1', 'a, 3'])
type(rdd)
处理地图和缩减功能的代码：
priceMap= s.map(lambda o: (o.split(",")[0], float
浏览 1提问于2020-11-12得票数 0
回答已采纳
1回答
我有以下RDD (名称: AllTrainingDATA_RDD)，它的类型是AllTrainingDATA_RDD，：
(ICCH_1,4.3,3.0,Iris-setosa)
(ICCH_1,4.4,2.9,Iris-setosa)
(ICCH_1,4.4,3.0,Iris-setosa)
(ICCH_2,4.4,3.2,Iris-setosa)
第1栏: ICCH_ID，第2栏: X_Coordinates，第3栏: Y_Coordinates，第4栏:类
我想以一个RDD结束，其中第2和第3列作为键，第4列作为值。列ICCH_ID应该保留在RDD中。
我目前基于一些互联网研究的尝试是这
浏览 0提问于2019-07-20得票数 1
回答已采纳
我正在尝试使用map()执行列转换。然而，它不适用于以下一栏：-
from pyspark.sql import SparkSession
data = [('James','Smith','M',30),
('Anna','Rose','F',41),
('Robert','Williams','M',62)]
rdd = spark.sparkContext.parallelize(data)
columns = ["firstn
浏览 7提问于2022-07-14得票数 0
回答已采纳
我是很新的火花编程。我正在尝试将地图和reduceByKey实现到以下15个字段的数据集。
rdd=sc.parallelize([
("West", "Apple", 2.0, 10,2.0, 10,2.0, 10,2.0, 10,2.0, 10,2.0,2.0, 10),
("West", "Apple", 3.0, 10,2.0, 10,2.0, 10,2.0, 10,2.0, 10,2.0, 10,2.0])
这是我的map函数，在这里我试图创建一个包含多个键和值的元组。
rdd1 = rdd.map(lambda
浏览 1提问于2015-08-27得票数 1
我正在尝试使用Apache-Spark来并行处理目录中分离的csv文件。具体地说，我希望每个从节点将每个csv文件中第一列的所有编号相加，并返回计算结果。以下是我的代码：
import os, sys, inspect, csv
### Current directory path.
curr_dir = os.path.split(inspect.getfile(inspect.currentframe()))[0]
### Setup the environment variables
spark_home_dir = os.path.realpath(os.path.abspath(
浏览 1提问于2015-07-15得票数 1
1回答
我有一个作为元组对列表的RDD：
[((1.1, 1.2), (0.1, 0.1)),
((1.1, 1.2), (0.4, 2.1)),
((0.1, 0.1), (1.1, 1.2)),
((9.0, 9.0), (9.1, 9.1)),
((0.4, 2.1), (1.1, 1.2)),
((9.1, 9.1), (9.0, 9.0))]
如何从该RDD中删除重复的元组，并获得所需的输出：
[((1.1, 1.2), (0.1, 0.1)),
((1.1, 1.2), (0.4, 2.1)),
((9.1, 9.1), (9.0, 9.0))]
我已经尝试了distinct
浏览 5提问于2018-02-27得票数 0
回答已采纳
我在RDD上遇到了glom()方法。根据文件
返回通过将每个分区中的所有元素合并为数组而创建的RDD。
glom是在分区间对数据进行洗牌，还是只将分区数据作为数组返回？在后一种情况下，我认为可以使用mapPartitions实现同样的目标。
我还想知道是否有任何从glom中受益的用例。
浏览 4提问于2016-03-02得票数 15
回答已采纳
我的场景是从现有的RDD捕获一些(不是全部)数据，然后将其传递给其他Scala class进行实际操作。让我们看看文本文件中的示例数据(empnum，empname，emplocation，empsal)。
11,John,Paris,1000
12,Daniel,UK,3000
第一步，我用下面的代码用RDD[String]创建一个RDD，
val empRDD = spark
.sparkContext
.textFile("empInfo.txt")
因此，我的要求是使用empnum、empname、emplocation (同样使用RDD[String])创建
浏览 17提问于2020-05-08得票数 2
回答已采纳
1回答
我有一个RDD的表格：
(2，你好，嗨，你好，你好，你)
我需要把这些元组映射成：
((2，你好)，(2，嗨)，(2，如何)，(2，是)，(2，你))
我正在用python尝试这个：
PairRDD = rdd.flatMap(lambda (k,v): v.split(',')).map(lambda x: (k,x)).reduceByKey())
这将无法工作，因为我在地图转换中没有k。我不知道怎么做？有什么评论吗？
提前感谢你。
浏览 5提问于2015-11-06得票数 0
我有一个案例类(TopNModel)的RDD，并且希望从通过tx + rx排序的RDD中获得顶级N个元素。在两个相等(tx + rx)按mac排序的情况下。
case class TopNModel(mac: Long, tx: Int, rx: Int)
例如：
RDD[TopNModel(10L, 200, 100), TopNModel(12L, 100, 100), TopNModel(1L, 200, 400), TopNModel(11L, 100, 200)]
按tx + rx和mac进行排序：
RDD[TopNModel(1L, 200, 400), TopNModel(10L
浏览 3提问于2017-12-04得票数 0
我的初始RDD如下所示：
RDD_int.collect() = [("a", 0, LabeledPoint(0, f01, f02, ..., f0n)), ("a", 1, LabeledPoint(0, f11, f12, ..., f1n)), ..., ("a", m, LabeledPoint(0, fm1, fm2, ..., fmn))]
我尝试这个操作：
RDD.filter(lambda x : "LabeledPoint" in x[2])
它不起作用。
我希望得到这一条：
RDD.collect() =
浏览 0提问于2021-05-01得票数 1
我已经扩展了现有的WordCount示例，并在Apache的官方站点上解释了这个例子。延期如下：
根据元组的存在性对元组进行排序。例如:现有的未排序顺序：(nec,8)，(eu,11)，(sit,7)我想要的排序顺序：(eu,11)，(nec,8)，(sit,7)
从排序列表中拿出前三名。排序做得很好，但是top()不在JavaPairRDD上工作。让我粘贴我的代码。其他方法是相同的，所以我在这里写我的主要方法：
公共静态空主(String[] args) { if (args.length < 1) { System.err .println(“请提供输入文件完整路径
浏览 3提问于2015-06-23得票数 0
1回答
下面有来自rdd是RDD[(String,Vector)]的代码片段，但不幸的是，我的Scala编译器在调用flagVectorOutlier(rdd, predictedRDD)的错误Type mismatch, expected: RDD[(String,Vector)], actual: RDD[(String,Vector)]中抱怨
def someFunction() {
testData.foreachRDD( rdd => {
val vectorsRDD = rdd.map( pair => pair._2 )
val predictedRD
浏览 2提问于2015-03-10得票数 1
我创建的rdd如下所示
rdd=sc.parallelize([('AA', 44),('BB', 53),(('AA', 'Bb'), 23), (('AD', 'AC'), 23),(('AA', 'BB', 'CC'), 2)])
我想把元组关键字转换成字符串。
我的预期输出如下new_rdd.collect()应该给出的：
[('AA', 44),('BB', 53),('AA,Bb', 23),
浏览 93提问于2018-12-07得票数 -2
回答已采纳
1回答
我试图在PySpark中实现一个简单的GloVe算法，它使用的是来自的预先训练的GloVe模型。
我有两个RDDs：
在(K:V)形式中名为documents的一对RDD，其中K是文档ID，V是该文档中所有单词的列表，例如('testDoc1':'i am using spark') ('testDoc2':'testing spark')
一个名为words的对RDD，它以K:V的形式表示单词嵌入，其中K是一个单词，V是表示这个单词的向量，例如('i', [0.1, 0.1, 0.1]) (&
浏览 1提问于2018-02-28得票数 3
回答已采纳
1回答
我正在尝试将下面的流水线RDD转换为dataframe。
流水线RDD -> user_rdd
['new_user1',
'new_user2',
'Onlyknows',
'Icetea',
'_coldcoffee_']
我试图使用以下代码进行转换
schema = StructType([StructField('Username', StringType(), True)])
user_df = sqlContext.createDataFrame(user_rdd,sche
浏览 4提问于2017-12-29得票数 0
回答已采纳
1回答
我正在尝试创建一个kafka流，然后对其进行一些转换，但我创建的流似乎是空的。我将一个文本文件加载到producer中，然后通过consumer使用它，它工作得很好，但它不会创建kafka流。包含36000个条目的输入文本文件如下所示：
10.000000
26.000000
-8.000000
-28.000000
...
我的python代码是：
sc = SparkContext(appName="STALTA")
ssc = StreamingContext(sc, 2)
broker, topic = sys.argv[1:]
kvs = KafkaUtils.cre
浏览 2提问于2018-07-17得票数 0
我目前正尝试在星火集群上执行LDA。我有一个RDD
>>> myRdd.take(2)
[(218603, [0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0]), (95680, [0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0])]
但呼唤
model = LDA.train(myRdd, k=5, seed=42)
从工作人员中提供下列错误：
org.apache.spark.SparkException:由于阶段失败而中止的任务:阶段5874.0中的任务0失败了4次，最
浏览 3提问于2016-11-08得票数 1
回答已采纳
2回答
在为PySpark的max函数提供的示例中：
火花缭乱
>>> rdd = sc.parallelize([1.0, 5.0, 43.0, 10.0])
>>> rdd.max()
43.0
>>> rdd.max(key=str)
5.0
Q1。当key=str时，如何将其评估为5.0？它是否基于对字符/字符串类型的转换？
Q2。参数"key“可以取的值是多少？
我还在这个位置找到了"max“的函数定义。
https://github.com/adobe-research/spark-gpu/blob/master/src
浏览 0提问于2016-04-19得票数 0
回答已采纳
1回答
假设我有一个具有以下类型的RDD：
RDD[(Long, List(Integer))]
我是否可以假设整个列表位于同一个员工的位置？我想知道某些操作在RDD级别上是可以接受的，还是应该在驱动程序中计算。例如：
val data: RDD[(Long, List(Integer))] = someFunction() //creates list for each timeslot
请注意，List可能是aggregate或任何其他操作的结果，而不一定是作为一个整体创建的。
val diffFromMax = data.map(item => (item._1, findDiffFrom
浏览 3提问于2017-01-02得票数 0
回答已采纳
1回答
考虑到这样的数据：
val my_data = sc.parallelize(Array(
"Key1, foobar, 10, twenty, 20",
"Key2, impt, 11, sixty, 6",
"Key3, helloworld, 110, seventy, 9"))
我想过滤并创建一个key,value RDD，如下所示：
key1, foobar
key1, twenty
key2, impt
key2, sixty
key3, helloworld
key3, seventy
我试过什么
我想我可以把数据放在一
浏览 4提问于2017-09-15得票数 2
回答已采纳
1回答
我不是RDD方面的专家，我想找一些答案来解决这个问题，我试着在pyspark上执行很少的操作，但是无法实现，特别是通过子处理，我知道我可以通过将RDD转换成DF来实现这一点，但是我想知道在之前的DF时代之前是如何做到的？公司是否仍然更喜欢在RDD或dataframes中工作？
我的代码：
rdd= sc.textFile("Sales.txt")
##Taking only required columns and changing the data types
rdd_map = rdd.map(lambda line: (int((line.split("|"
浏览 0提问于2019-08-17得票数 0
回答已采纳
1回答
我一直在尝试将RDDs转换为DataFrames格式，然后再转换回来。首先，我有一个类型为(Int，Int)的RDD，名为dataPair。然后，我使用以下命令创建了一个带有列标题的DataFrame对象：
val dataFrame = dataPair.toDF(header(0), header(1))
然后，我使用以下命令将其从DataFrame转换回RDD：
val testRDD = dataFrame.rdd
它返回org.apache.spark.sql.Row类型的RDD (不是(整数，整数))。然后我想用.toDF把它转换回RDD，但是我得到了一个错误：
error: va
浏览 1提问于2016-05-04得票数 11
回答已采纳
我不确定标题是否准确地描述了我的问题，但以下是我的问题：
dump类型为：
dump: org.apache.spark.rdd.RDD[(String, String, String, String)]
例如：
val dump = sc.parallelize(List(("a","b","c","s")))
我有以下的for-循环：
for (i <- List(0,1,2,3)) {
val temp = dump.map(x=> x._i)
}
但是IntelliJ表示x._i中有一
浏览 0提问于2015-10-11得票数 0
回答已采纳
我有一个在数据中有一些空值的RDD[(Seq[String], Seq[String])]。转换为dataframe的RDD如下所示
+----------+----------+
| col1| col2|
+----------+----------+
|[111, aaa]|[xx, null]|
+----------+----------+
以下是示例代码：
val rdd = sc.parallelize(Seq((Seq("111","aaa"),Seq("xx",null))))
val df = rdd.to
浏览 2提问于2017-04-06得票数 3
我遵循中的说明。
import sqlContext.implicits._
val people: RDD[Person] = ...
people.saveAsParquetFile("people.parquet")
下面是我的代码：
val rawRDD = sc.textFile(peopleFile).map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt))
Person，这里是case class。
当我用
rawRDD.saveAsParquetFile("peopl
浏览 3提问于2015-04-14得票数 0
1回答
我有一个星火流，记录在其中流动。间隔大小是1秒。我想把流中的所有数据结合起来。因此，我创建了一个空的RDD，然后使用transform方法与这个空的RDD进行RDD的合并(在流中)。我期望这个空的RDD在结束时拥有所有的数据。但是这个RDD始终是空的。
还有谁能告诉我我的逻辑是否正确。
JavaRDD<Row> records = ss.emptyDataFrame().toJavaRDD();
JavaDStream<Row> transformedMessages = messages.flatMap(record -> processData(record)
浏览 6提问于2017-06-12得票数 0
回答已采纳
1回答
我有一个RDD，其中我的键是一个id，值包括id列表。例如，我想按升序排序值列表。
1, list(12,3,8,10)
2, list(42,3,65,33)
3, list(6,2,4,1)
输出
1, list(3,8,10,12)
2, list(3,33,42,65)
3, list(1,2,4,6)
RDD创建，所以我在加入两个不同的RDD之后创建了RDD，然后使用它productIterator来创建list of values，这给出了RDD类型的RDD(Int, List[Any])。
我尝试过rdd.mapValues(x=> _.2.sorted)不同的排序方法，但
浏览 1提问于2019-05-21得票数 0
我正在尝试向Scala2.11中的ConcurrentLinkedQueue (collectedData)添加一个数组：
stream.foreachRDD { rdd => Collections.addAll(collectedData, rdd.collect()) }
我得到了以下编译错误：
[error] found : java.util.concurrent.ConcurrentLinkedQueue[(String, String)]
[error] required: java.util.Collection[_ >: java.io.Serializa
浏览 1提问于2016-02-06得票数 0
我有两个rdd，我想对rdd1的每个项进行一些rdd1项的计算。因此，我在用户定义的函数中传递RDD2，如下所示，但是我得到了类似于rdd1 cannot be passed in another rdd的错误。如果我想在两个rdd上执行操作，我可以知道如何实现这个目标吗？
例如：
RDD1.map(line =>function(line,RDD2))
浏览 3提问于2016-01-16得票数 5
我正试着从spark给redis写信。但是我得到了一个编译时错误，说“在redis.clients.jedis.Jedis中不能访问类BinaryJedis中的变量管道”。我的代码如下(部分显示)：
import org.sedis._
import redis.clients.jedis._
...
val myRDD = KafkaUtils.createStream(ssc, zkQuorum, group, topic).map(_._2).window(Seconds(300), Seconds(10))
myRDD.foreachRDD( r
浏览 0提问于2015-12-13得票数 0
我的RDD all_keys存储值列表：
>> all_keys.take(3)
[['a','b','c'],
['a','b'],
['a','d','f']]
大多数这些值都是重复出现的，但有些列表中的值在其他列表中不存在。
如何从存储在RDD中的所有列表中获取唯一元素列表？
浏览 0提问于2016-06-23得票数 0
回答已采纳
1回答
创建映射的RDD并将其保存为文本
、、、、
我运行了一个K-表示示例，我有一个RDD，我的数据名为parsedData，我的模型名为集群。我想从模型中创建一个带有数据池和预测集群的映射Rdd。所以我试着
val predictions = parsedData.map( point =>
{
val pointPred = clusters.predict(point)
Array(point,pointPred)
})
当我尝试
predictions.first()
我拿着
ArrayAny = Array(0.8898668778942382,0.89533945283595，0)
这就是我想要的结果。所以我试
浏览 3提问于2017-09-08得票数 2
回答已采纳
2回答
我的输入行如下所示
t1, file1, 1, 1, 1
t1, file1, 1, 2, 3
t1, file2, 2, 2, 2, 2
t2, file1, 5, 5, 5
t2, file2, 1, 1, 2, 2
我想要像下面的行那样实现输出，这是对应数字的垂直加法。
file1 : [ 1+1+5, 1+2+5, 1+3+5 ]
file2 : [ 2+1, 2+1, 2+2, 2+2 ]
我处于星星之火的流媒体环境中，我很难找到按文件名进行聚合的方法。
似乎我需要使用如下所示，我不知道如何获得正确的语法。任何投入都是有帮助的。
myDStream.foreachRDD(rdd =&
浏览 3提问于2016-02-21得票数 1
回答已采纳
我在PySpark中有两个PySpark：
RDD1:
[(u'2013-01-31 00:00:00', u'a', u'Pab', u'abc', u'd'),(u'2013-01-31 00:00:00', u'a', u'ab', u'abc', u'g'),.....]
RDD2:
[(u'41',u'42.0'),(u'24',u'98.0'),....]
两个
浏览 7提问于2015-12-07得票数 5
我有以下rdd
[('K', ' M'),
('K', ' H'),
('M', ' K'),
('M', ' E'),
('H', ' F'),
('B', ' T'),
('B', ' H'),
('E', ' K'),
('E', ' H'),
('F', ' K
浏览 2提问于2020-06-29得票数 0
回答已采纳
我是Python新手。
Scala代码：
rdd1为字符串格式。
rdd1=sc.parallelize("[Canada,47;97;33;94;6]", "[Canada,59;98;24;83;3]","[Canada,77;63;93;86;62]")
val resultRDD = rdd1.map { r =>
val Array(country, values) = r.replaceAll("\\[|\\]", "").split(",")
country
浏览 3提问于2019-11-25得票数 2
回答已采纳
我很抱歉，这可能是一个简单的问题，但我很难理解如何用scala/spark来解析rdd。我有一个从CSV创建的RDD，用
val partitions: RDD[(String, String, String, String, String)] = withoutHeader.mapPartitions(lines => {
val parser = new CSVParser(',')
lines.map(line => {
val columns = parser.p
浏览 0提问于2015-03-26得票数 0
回答已采纳
我刚接触过火花和转化，所以需要一点帮助。我设法找到了相反的解决方案。当你把一个值变成一个列表而不是从。我想“压平”我猜钥匙的价值。我有一个类似于Map的RDD，其中键是Int，值是String的列表。我需要用键和列表中的每个值对到一个RDD中。
浏览 3提问于2017-11-24得票数 1
回答已采纳
我正在处理Apache日志文件。我使用每个日志行中的元组(day，host)创建了RDD。下一步是对主机进行分组，然后显示结果。
我使用distinct()将第一个RDD映射为(day，host)元组。当我不使用distinct时，我会得到与使用时不同的结果。那么当在spark中使用distinct()时，结果会发生怎样的变化呢？
浏览 4提问于2015-06-17得票数 0
我想将流数据插入到hbase中；这是我的代码：
val tableName = "streamingz"
val conf = HBaseConfiguration.create()
conf.addResource(new Path("file:///opt/cloudera/parcels/CDH-5.4.7-1.cdh5.4.7.p0.3/etc/hbase/conf.dist/hbase-site.xml"))
conf.set(TableInputFormat.INPUT_TABLE, tableName)
val admin = new HBase
浏览 0提问于2017-04-20得票数 0
Description .如何在DataFrame中将字典数据集转换为PySpark：
错误/未预期结果--我试过了
df = spark.createDataFrame([Row(**i) for i in bounds])
但是得到：
TypeError:无法推断类型的架构：<class 'str'>
此代码：
rdd = sc.parallelize(bounds)
rdd.map(lambda x: (x,)).toDF().show()
而其他人则给出了意想不到的结果。
预期结果
My DataSet
浏览 1提问于2021-12-04得票数 1
回答已采纳
因此，在Spark中，当一个应用程序启动时，就会创建一个包含该应用程序的数据集(例如，WordCount的words数据集)的RDD。
到目前为止，我所理解的是，RDD是WordCount中这些单词的集合，以及对这些数据集所做的操作(例如，map、reduceByKey等)。
然而，afaik，Spark也有HadoopPartition (或一般的:分区)，它由每个执行器从HDFS读取。我相信驱动程序中的RDD也包含所有这些分区。
那么，Spark中的执行者之间是如何划分的呢？是否每个执行器都将这些子数据集作为单个RDD获取，与驱动程序中的RDD相比，RDD包含的数据更少，还是每个执行器只处
浏览 1提问于2016-04-11得票数 0
1回答
我使用的是一个RDD，它的结构是这样的：[Int，ListInt]我的目标是用键映射每个对的列表中的项目。例如，我需要这样做：
RDD1:[Int, List[Int]]
<1><[2, 3]>
<2><[3, 5, 8]>
RDD2:[Int, Int]
<1><2>
<1><3>
<2><3>
<2><5>
<2><8>
嗯，我不知道要到RDD2需要什么样的转换。转换列表可以找到。知道吗？这是错误的做法吗？
浏览 1提问于2017-03-17得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
