我正在使用Spark RDD API收集数据，并创建了一个成对的RDD，如下所示： spark = SparkSession.builder.master('local').appName('app'.map(lambda x: x.split(","))\
.map(lambda x: (x[2], [x[1], x[3],x[5]])) 以下是成对RDD的示例摘录： [('
浏览 126提问于2021-10-14得票数 1
回答已采纳
我想为一个大的spark数据帧计算成对的kendall的tau等级相关性。它很大(比如10m行，10k列)，不能转换成pandas数据帧，然后使用pandas.DataFrame.corr进行计算。此外，每列可能具有空值，因此在计算成对的kendall's tau时，需要排除两列中任何一列中具有空值的行。
我查过pyspark.mllib.stat.Statistics.corr了。df_rdd</
浏览 18提问于2019-07-20得票数 2
2回答
spark = SparkSession.builder.getOrCreate()ssc = StreamingContext(sc , 10)rdd.take(1)[['0.02703300', '1.30900000'],0.02704600', '3.90800000'],
['0
浏览 6提问于2021-01-02得票数 1
1回答
我正在使用spark DataFrame，我希望检测来自特定列的任何值，其中该值不是单调递减的。对于这些值，我想根据排序条件将它们替换为以前的值。下面是一个概念性的示例，假设我有一个值为[65, 66, 62, 100, 40]的列。值"100“不遵循单调下降趋势，因此应替换为62。因此，结果列表将是[65, 66, 62, 62, 40]。下面是我创建
浏览 15提问于2020-02-14得票数 1
回答已采纳
2回答
我正在处理将SQL代码转换为PySpark代码，并遇到了一些SQL语句。我不知道如何在pyspark中使用case语句？我计划创建一个RDD，然后使用rdd.map，然后做一些逻辑检查。这是正确的方法吗？请帮帮我！ case
when (e."
浏览 2提问于2016-10-12得票数 24
1回答
如何在RDD(key，value)中迭代值。tsRDD.map(lambda x:(x,1)).groupByKey()[('abc', <pyspark.resultiterable.ResultIterable object at 0xb0e8242c>), ('xyz', <<
浏览 1提问于2015-06-30得票数 2
1回答
我有以下的pyspark RDD及其I和计数：id_list = ['12', '125', '78']
我想要一个新的键列表，值</
浏览 8提问于2021-03-13得票数 1
回答已采纳
1回答
我有重复的行，可能包含相同的数据或在PySpark数据框中有缺失值。我写的代码非常慢，并且不能作为分布式系统工作。有谁知道如何从PySpark数据帧中的重复行中保留单个唯一值，该数据帧可以作为分布式系统运行，并且具有快速的处理时间？ 我已经写了完整的Pyspark代码，这个代码工作正常。():
# Match duplicates using std na
浏览 25提问于2019-06-21得票数 0
我正在创建一个spark，其中模式是从json records.But推断出来的，其中一些json数据集的行比其他行有更多的列，因此数据格式解析失败。我是否可以将空值替换为缺少的额外列的记录。raw_event_data_rdd = sc.textFile(INPUT_DATA_DIR)
pre_processed_raw_event_data_rdd = raw_event_data_rdd.ma
浏览 2提问于2017-10-31得票数 0
1回答
在下面的代码中，我希望看到初始的RDD，就像在函数myfunc中一样，我只是在打印值之后返回迭代器。但是，当我在RDD上执行collect时，它是空的。from pyspark import SparkConf print(it.next()) n = 5
rdd</e
浏览 0提问于2017-03-23得票数 1
回答已采纳
我有两个((a, 0), (b, 1), (c, 2))，一个是(a, b, a, c, b, c, a)，另一个是成对的RDD。我想将第一个RDD中的as，bs和cs分别替换为0,1,2 (它们分别是第二个RDD中的键a，b，c的值)。我想保留第一个RDD中事件的顺序。
如何在Spark中实现它？
浏览 4提问于2016-01-05得票数 1
1回答
我在脚本中使用for循环为size_DF(数据帧)的每个元素调用一个函数，但这需要很长时间。我尝试通过逐个映射删除for循环，但是我没有得到任何输出。size_DF是我从表中获取的大约300个元素的列表。用于：size_RDD = sc.parallelizelength, end_date)if len(size_DF) == 0:
print "
浏览 0提问于2018-02-28得票数 2
回答已采纳
2回答
我有很大的时间序列数据，数据格式是：(arrival_time，键，值)，时间单位是秒，例如：0.03, k, v1.00, k, v1.20我需要做的是获取整个数据每秒的行数。到目前为止，我使用的是pySpark，我的代码如下：lo = rdd.take(1)[0]end = rdd.collect()[-1]
浏览 1提问于2015-11-16得票数 4
回答已采纳
我有这样的DenseVector RDD[DenseVector([1.0, 0.0, 1.0, 1.0, 0.0, 0.0,我试过像这样它会产生这样的错误
Traceback/sql/session.py", line 520,
浏览 3提问于2016-12-26得票数 11
回答已采纳
1回答
加入PySpark不加入任何值
、、、、
在PySpark中，我想用键值对对两个RDD进行完全的外部连接，其中键可以是None。])看起来，PySpark加入了键为空的记录：>>> [(None, ('a',a NULL c
SEL
浏览 0提问于2016-03-02得票数 1
我使用Pyspark来编写这段代码： df.na.fill("").show()rdd2=df.rdd.map(lambda x:(x.firstName+""+x.lastName,x.street+","+x.town,x.city,x.code) #error linedf2=rdd2.toDF(["name，上面写着： PythonException: 'Typ
浏览 16提问于2021-09-28得票数 0
回答已采纳
在PySpark (v1.6.2)中，当使用指定的模式将RDD转换为DataFrame时，值类型与模式中声明的值类型不匹配的字段将转换为null。from pyspark import SparkContextfrom pyspark.sql.types import StructType= sc.parallelize([{"foo&
浏览 27提问于2016-07-27得票数 0
回答已采纳
我需要从另一个成对的RDD生成成对的RDD。基本上，我正在尝试编写一个执行以下操作的map函数。RDD[Polygon,HashSet[Point]] => RDD[Polygon,Integer]Scala函数，遍历HashSet并将"Point“对象的值相加。._1,Integer.valueOf(outageCnt))
浏览 6提问于2017-07-11得票数 0
回答已采纳
1回答
我正在构建一个解析器，它接受"key"="value“对的原始文本文件，并使用PySpark写入tabular/..csv结构。在我被困的地方，我可以访问函数中的键和值来构造每个csv_row，甚至可以检查键是否等于预期键(col_list)的列表，但是当我在lambda中调用函数processCsv时，我不知道如何将每个csv_row如何以键/值格式遍历RDD<em
浏览 5提问于2017-08-02得票数 1
回答已采纳
云服务器
ICP备案
实时音视频
对象存储
云直播
腾讯云开发者
