Pyspark:按键聚合RDD，然后也按键对元组值列表求和
Pyspark是一个基于Python的Spark编程接口，它提供了一种方便的方式来处理大规模数据集。在Pyspark中，按键聚合RDD是一种常见的操作，它可以将具有相同键的元素聚合在一起，并对相应的值列表进行求和。
按键聚合RDD的过程如下：
- 创建一个包含键值对的RDD。
- 使用reduceByKey()函数按键对值进行聚合。该函数将具有相同键的元素聚合在一起，并对相应的值列表进行操作（例如求和、求平均等）。
- 返回一个新的RDD，其中包含按键聚合后的结果。
按键聚合RDD的优势：
- 高效性：Pyspark使用分布式计算框架Spark，可以在集群上并行处理大规模数据集，提高计算效率。
- 灵活性：按键聚合RDD可以根据具体需求进行不同的聚合操作，例如求和、求平均、计数等。
- 可扩展性：Pyspark可以处理大规模数据集，并且可以通过添加更多的计算节点来实现横向扩展。
按键聚合RDD的应用场景：
- 数据分析：按键聚合RDD可以用于对大规模数据集进行统计分析，例如计算每个键对应的平均值、求和等。
- 日志处理：按键聚合RDD可以用于对日志数据进行分析，例如按照时间戳对日志进行聚合，统计每个时间段的访问量等。
- 推荐系统：按键聚合RDD可以用于对用户行为数据进行分析，例如统计每个用户的购买次数、评分等。
腾讯云相关产品推荐：
- 腾讯云Spark：腾讯云提供的Spark服务，可以方便地进行大规模数据处理和分析。详情请参考：腾讯云Spark产品介绍
请注意，以上答案仅供参考，具体的产品选择和推荐应根据实际需求和情况进行评估。
相关·内容
在scala中，当我有一个类似于：List(("a",1),("a",2),("b",3),("b",4),("b",5),("a",6))的RDD列表时，我想要计算每个字符的avg数。
与a一样，它以1+2+6 = 9的值显示了3次，因此我希望得到的结果是(a, 3)。
在Scala中，我可以编写如下代码：
val newRdd = rdd.aggregateByKey((0,0))((t,v) => {(t._1 + v, t._2 + 1)}, (t1, t2) => {(t1._
浏览 5提问于2021-12-24得票数 0
回答已采纳
我从这个网站下载了一个文本文件：http://snap.stanford.edu/data/web-Amazon-links.html，目的是在Pyspark中做一些文本分析。 所以我设置了我的spark上下文： from pyspark import SparkConf, SparkContext
conf = SparkConf().setAppName('app')
sc = SparkContext(conf=conf)
from pyspark.sql import SQLContext 我抓起文件： Data1 =sc.textFile('/home/jo
浏览 44提问于2019-04-02得票数 1
回答已采纳
我是Spark (使用1.1版本)和Scala的新手。我正在转换我现有的Hadoop MapReduce代码，以火花MR使用Scala和位丢失。
我希望映射的RDD按键分组。当我在网上读到的时候，有人建议我们应该避免使用groupByKey，而使用reducedByKey。但是，当我应用reduceBykey时，我并没有像我的代码=>Ex所期望的那样，得到给定键的值列表。
val rdd = sc.parallelize(List(("k1", "v11"), ("k1", "v21"), ("k2", &#
浏览 1提问于2016-01-18得票数 1
我有一个pyspark，它可以收集成一个元组列表，如下所示：
rdds = self.sc.parallelize([(("good", "spark"), 1), (("sood", "hpark"), 1), (("god", "spak"), 1),
(("food", "spark"), 1), (("fggood", "ssspark"), 1), (("
浏览 2提问于2020-06-17得票数 2
回答已采纳
我正在查找数据中所有列的“模式”，并将它们存储在列表中。计算每个列的我的模式的代码：
from pyspark.sql.functions import *
#calculating mode value
mode_val = []
for i in df_num.columns :
cnts = df_num.groupBy(i).count()
mode = cnts.join(
cnts.agg(max("count").alias("max_")), col("count") == col(&#
浏览 0提问于2018-04-03得票数 2
回答已采纳
我正在尝试运行中给出的Spark / Python的Logistic回归示例，并且已经成功地使用了Spark1.6和Python2.7。
现在我必须将它移到Spark2.1和Python3.5( 3.6是不兼容的)，我正在使用Ubuntu16.04中的木星笔记本
这段代码工作正常
# Evaluate the model on training data
labelsAndPreds = modelInput.map(lambda p: (p.label, LRmodel.predict(p.features)))
print(labelsAndPreds.count())
print(lab
浏览 2提问于2017-04-12得票数 1
回答已采纳
我目前正在自学Spark programming，并试图用PySpark重新编写一个现有的Python应用程序。然而，我仍然对如何在PySpark中使用常规Python对象感到困惑。
我了解Spark中的分布式数据结构，如RDD、DataFrame、Datasets、vector等。Spark有自己的转换操作和动作操作，如.map()、.reduceByKey()来操作这些对象。但是，如果我在PySpark中创建传统的Python数据对象，比如数组、列表、元组或字典，会怎么样呢？它们将只存储在我的驱动程序节点的内存中，对吗？如果我把它们转换成RDD，我还能用典型的Python函数做操作吗？
如
浏览 34提问于2017-03-01得票数 1
回答已采纳
2回答
使用Pyspark，我正在尝试使用RDD来基于该RDD的内容进行聚合。
我的RDD目前看起来像(显然有更多的数据)：
[([u'User1', u'2'], 1), ([u'User2', u'2'], 1), ([u'User1', u'3'], 1)]
我想将其聚合到以下格式中：
User1 5
User2 2
我正在努力与RDD交互，特别是RDD中的列表，以获取这些数据。我还希望将其保留为RDD，而不是将其转换为数据帧。
有谁能教我怎么做吗？
浏览 20提问于2021-02-17得票数 0
回答已采纳
1回答
我有一个DataFrame (转换为RDD)，并希望重新分区，以便每个键(第一列)都有自己的分区。这是我所做的：
# Repartition to # key partitions and map each row to a partition given their key rank
my_rdd = df.rdd.partitionBy(len(keys), lambda row: int(row[0]))
但是，当我试图将它映射回DataFrame或保存它时，我得到了这个错误：
Caused by: org.apache.spark.api.python.PythonException:
浏览 0提问于2015-11-21得票数 5
我有这样的数据：
+---+-------+--------------+
| id|clicked| features |
+---+-------+--------------+
| 7| 1.0 |(4,[3],[1.0]) |
| 8| 0.0 |(4,[2],[12.0])|
| 9| 0.0 |(4,[3],[0.1]) |
+---+-------+--------------+
并将其转换为RDD：
Row(id=7, clicked=1.0, features=SparseVector(4, {3: 1.0}))
Row(id=8, clicked=
浏览 3提问于2018-09-19得票数 2
回答已采纳
2回答
尝试将一些pySpark转换为scala等效项，而对于双列表理解的正确语法，我遇到了问题。代码接受键值列表，并返回为同一键发生的元组形式的值列表。意思是(2，(‘user3 1’，‘user3 2’，‘user3’)将返回(‘user3 1’，'user2')，(‘user3 1’，'user3')，(‘user3 2’，‘user3’)。
#source rdd
[(2, ['user1', 'user3']), (1, ['user1', 'user2', 'user1']),
浏览 2提问于2016-02-11得票数 1
回答已采纳
我有一个字典的RDD，我想得到一个只包含不同元素的RDD。但是，当我试图打电话给
rdd.distinct()
PySpark给出了以下错误
TypeError: unhashable type: 'dict'
at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)
at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)
at org.ap
浏览 5提问于2016-02-19得票数 2
回答已采纳
我有一个具有大量KeyVal条目的RDD。相同的密钥将出现多次，我感兴趣的是为每个键提取第一个N个条目。作为星火的新手，到目前为止，我还不知道该怎么做，所以我会很感激你的帮助。
输入可能类似于：
rdd = sc.parallelize([('a',1),('a',2),('b',3),('a',5),('b',4),('b',6)])
例如，每个键的前两个条目所需的最终输出：
output = {'a':[1,2], 'b':[3,4]}
如果我只对最前面的n个条目
浏览 1提问于2019-06-16得票数 1
回答已采纳
1回答
Jupyter PySpark发送错误=> TypeError：()缺少1个必需的位置参数：'y‘ 我正在使用Jupyter中的PySpark，并且有以下代码，它会向我发送以下错误： l = [i for i in range (0,3000)]
rdd = sc.parallelize(l) def check(x,y,k):
if (((2*x+1)**2)+((2*y+1)**2))<(2*k)**2:
return 1
else:
return 0 rdd4 = rdd.cartesian(rdd) rdd5 = r
浏览 12提问于2019-10-13得票数 0
回答已采纳
我已经为这件事摇头了一段时间了--我真的很感激你的建议！不好意思标题太长了，我希望下面的一个简短的例子能更好地解释这个问题。
假设我们有以下形式的RDD：
data = sc.parallelize([(1,[('k1',4),('k2',3),('k1',2)]),\
(2,[('k3',1),('k3',8),('k1',6)])])
data.collect()
输出：
[(1, [('k1', 4), ('k2', 3), ('k
浏览 0提问于2019-10-01得票数 1
回答已采纳
1回答
我想按值对K/V对进行排序，然后取最大的五个值。我设法做到了这一点，通过第一个映射恢复K/V，使用FALSE按降序排序，然后将key.value反向转换为原始的(第二个映射)，然后获取前5个最大的映射，代码如下：
RDD.map(lambda x:(x[1],x[0])).sortByKey(False).map(lambda x:(x[1],x[0])).take(5)
我知道在pySpark上有一个takeOrdered操作，但我只设法对值(而不是键)进行排序，我不知道如何进行降序排序：
RDD.takeOrdered(5,key = lambda x: x[1])
浏览 1提问于2015-06-12得票数 28
回答已采纳
我用Python编写了一个非常简单的火花代码：
import collections
Person = collections.namedtuple('Person', ['name', 'age', 'gender'])
a = sc.parallelize([['Barack Obama', 54, 'M'], ['Joe Biden', 74, 'M']])
a = a.map(lambda row: Person(*row))
print a.collec
浏览 5提问于2015-11-06得票数 2
回答已采纳
我一直试图阅读我的数据从一个卡夫卡的主题，并将它写到一个拼花文件。到目前为止，除了foreachRdd函数之外，一切都在工作。当我在dstream上使用映射时，我可以看到数据，但是在使用foreachRdd的下一步，Rdd总是空的，我不知道为什么。
我的环境是Ubuntu运行卡夫卡和火花独立。我用的是火花放电壳。我对python还不熟悉，所以在语法方面我仍然遇到了很多问题，我不知道这是否是我的问题所在。
任何帮助或洞察力都将不胜感激。
下面是我的代码的一个副本，我一直把它粘贴在pyspark外壳中
from pyspark import SparkContext
from pyspark.st
浏览 3提问于2018-01-07得票数 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
回答已采纳
有任何方法可以将一对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
回答已采纳
我正在使用以下代码将我的rdd转换为数据帧：
time_df = time_rdd.toDF(['my_time'])
并得到以下错误：
TypeErrorTraceback (most recent call last)
<ipython-input-40-ab9e3025f679> in <module>()
----> 1 time_df = time_rdd.toDF(['my_time'])
/usr/local/spark-latest/python/pyspark/sql/session.py in toDF(sel
浏览 0提问于2016-09-29得票数 2
我的rdd中的每个值都是一个元组：
temp = clustering.map(lambda x: (x[0][0], (1,1)))
temp.take(10)
[(0, (1, 1)),
(0, (1, 1)),
(6, (1, 1)),
(0, (1, 1)),
(0, (1, 1)),
(0, (1, 1)),
(0, (1, 1)),
(0, (1, 1)),
(7, (1, 1)),
(0, (1, 1))]
然后尝试通过key减少它：
temp.reduceByKey(lambda a,b: (a[1]+b[1])).collect()
并得到了这个错误：
浏览 0提问于2019-09-19得票数 0
我正在尝试使用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
我的train_rdd和(('a',1),('b',2),('c',3))一样。我使用以下方法将其转换为DataFrame
from pyspark.sql import Row
train_label_df = train_rdd.map(lambda x: (Row(**dict(x)))).toDF()
但也许某些RDDS中缺少了一些钥匙。所以错误就会发生。
File
"/mnt/hadoop/yarn/local/usercache/hdfs/appcache/application_/container_05_000017/py
浏览 8提问于2020-07-07得票数 1
回答已采纳
我有两个Spark1.4.1 PipelineRDD (我不知道这是什么对象:-s：
1) ids列表(ids_alsaciens RDD)
2)人员名单(人员RDD)
'Personnes‘RDD有4个字段，都是json格式，键是"id“。对于这个表中的同一个人，我可能有几行( id是相同的)。
我想获取'personnes‘RDD上的所有行，该id包含在'alsacien’表中。
我怎么能在火花中这样做呢？
>type(ids_alsaciens)
pyspark.rdd.PipelinedRDD
>type(personnes)
pyspark
浏览 0提问于2015-08-24得票数 1
回答已采纳
1回答
我想要找到这个查询的每月最大值，以获得每月最繁忙的机场。 在不使用SQL命令的情况下，是否有一个pyspark函数可以获得最大值并将RDD减少到只有12行(每月一行)？ from pyspark import SparkContext
from pyspark.sql import SQLContext
from operator import add
sc = SparkContext.getOrCreate()
sqlContext = SQLContext(sc)
rows = sc.textFile("data.csv")
data = rows.map(lamb
浏览 36提问于2020-06-17得票数 0
回答已采纳
3回答
我在理解可以调用像sortByKey这样的函数的集合的数据类型时遇到了问题。我有一个要调用sortByKey的ListMap (我想要一个按顺序存储元素的数据结构)。列表映射的内容是前n个数字作为关键字，1作为每个关键字的值。我应该使用什么来代替列表映射？
val l = (1 to 1000).toList
val d = ListMap(l.map(s=> s -> 1):_*)
val rdd = sc.parallelize(Seq(d))
rdd.collect()
val sorted = rdd.sortByKey()
sorted.collect()
浏览 2提问于2016-09-17得票数 0
有一个火花放电代码段。
seqOp = (lambda x,y: x+y)
sum_temp = df.rdd.map(lambda x: len(x.timestamp)).aggregate(0, seqOp, seqOp)
sum_temp的输出是一个数值。但我不清楚aggregate(0, seqOp, seqOp)是如何工作的。在我看来，通常情况下，aggregate只使用一个函数形式，比如"avg"。
此外，df.rdd.map(lambda x: len(x.timestamp))属于pyspark.rdd.PipelinedRDD型。我们怎样才能得到它的内容？
浏览 2提问于2021-04-27得票数 1
回答已采纳
我有一个pyspark RDD (myRDD)，它是一个可变长度的ID列表，比如
[['a', 'b', 'c'], ['d','f'], ['g', 'h', 'i','j']]
我有一个包含列ID和value的pyspark数据框架(myDF)。
我想用以下查询来查询myDF：
outputDF = myDF.select(F.collect_set("value")).alias("my_values").wher
浏览 26提问于2017-08-24得票数 0
我在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
我正在应用一个映射，然后使用pyspark在RDD上应用reduceByKey转换。我尝试了以下两种语法，这两种语法似乎都有效：
案例1：
my_rdd_out = my_rdd.map(lambda r: [r['my_id'], [[r['my_value']]]])\
.reduceByKey(lambda a, b: a+b)\
.map(lambda r: r[1])
案例2：
my_rdd_out = my_rdd.map(
浏览 0提问于2016-09-26得票数 0
回答已采纳
我似乎遵循了文档化的方式来显示从带有模式的RDD转换而来的DF。但很明显，我遗漏了一些很小但很重要的一点。然后如下：
# Original schema + Index for zipWithIndex with variations on this
schema = StructType(result_df.schema.fields[:] + [StructField("index", LongType(), True)])
rdd = result_df.rdd.zipWithIndex()
df = spark.createDataFrame(rdd, schema)
浏览 0提问于2019-08-12得票数 2
回答已采纳
1回答
在查看了星星之火/示例目录中的kmeans示例之后，我尝试在一组纬度和经度数据上进行K-意思聚类。我已经将.csv数据导入到spark (~1M行)中，并试图将数据读取作为我的k-means模型的输入，但是我一直得到一个错误。
我的星星之火看起来像：
ID col1 col2 Latitude Longitude
ford ... ... 22.2 13.5
landrover ... ...
浏览 3提问于2017-07-05得票数 1
回答已采纳
我需要在url中传递坐标，但我需要将rdd转换为字符串并用分号分隔。
all_coord_iso_rdd.take(4)
[(-73.57534790039062, 45.5311393737793),
(-73.574951171875, 45.529457092285156),
(-73.5749282836914, 45.52922821044922),
(-73.57501220703125, 45.52901077270508)]
type(all_coord_iso_rdd)
pyspark.rdd.PipelinedRDD
结果查找：
"-73.57534790
浏览 0提问于2018-04-12得票数 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
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
回答已采纳
我是pyspark的新手，我正在尝试将python中的列表转换为rdd，然后我需要使用rdd查找元素索引。对于第一部分，我正在做的是：
list = [[1,2],[1,4]]
rdd = sc.parallelize(list).cache()
所以现在rdd实际上是我的列表。问题是，我想要找到任意元素的索引，类似于" index“函数，它适用于python列表。我知道一个名为zipWithIndex的函数，它为每个元素分配索引，但我在python中找不到合适的示例(有java和scala的示例)。
谢谢。
浏览 2提问于2016-04-06得票数 6
回答已采纳
我的初始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
2回答
在我的应用程序中，当使用perfromance数字时，groupby占用了大量的时间。
我的RDD的结构如下：
JavaPairRDD<CustomTuple, Map<String, Double>>
CustomTuple:该对象包含有关RDD中当前行的信息，如星期、月份、城市等。
public class CustomTuple implements Serializable{
private Map hierarchyMap = null;
private Map granularMap = null;
private String timePeriod =
浏览 5提问于2015-01-16得票数 1
1回答
我有一个PySpark DataFrame，它有一个列作为一个热编码向量。我想在群后通过矢量加法来聚合不同的热编码向量。
例如df[userid,action] Row1: ["1234","[1,0,0]] Row2: ["1234", [0 1 0]]
我希望输出为row：["1234", [ 1 1 0]]，因此向量是由userid分组的所有向量的总和。
我怎样才能做到这一点？PySpark和聚合操作不支持向量加法。
浏览 4提问于2016-12-07得票数 12
我有一个具有以下(键、值)格式的RDD：
[((u'en', 1),[('term1', 2),('term2', 8),('term3', 6))]
注意到，这里的value是一个元组列表：(term,count)
我尝试使用以下方法按值对此rdd进行排序：
.map(lambda (a,b): (b,a)).sortByKey(False).map(lambda (a,b): (b,a))
但是它通过term和count产生了一个排序的RDD。
问题是如何通过count对这个RDD进行排序，从而产生以下结果：
[((u'
浏览 7提问于2016-01-05得票数 0
回答已采纳
我在使用星火的推荐系统。
在训练了一个模型之后，我做了下面的代码来获得推荐model.recommendProductsForUsers(2)
[(10000, (Rating(user=10000, product=14780773, rating=7.35695469892999e-05),
Rating(user=10000, product=17229476, rating=5.648606256948921e-05))),
(0, (Rating(user=0, product=16750010, rating=0.04405213492474741),
浏览 1提问于2016-03-28得票数 2
回答已采纳
我正尝试在我的服务器上运行一个简单的pandas UDF示例。来自
为了运行这段代码，我创建了一个全新的环境。
(PySparkEnv) $ conda list
# packages in environment at /home/shekhar/.conda/envs/PySparkEnv:
#
# Name Version Build Channel
arrow-cpp 0.10.0 py36h70250a7_0 conda-forge
blas
浏览 4提问于2018-09-14得票数 5
我试着用火花来实现一些种源算法。我想要做的第一步是将prov文档对象存储到spark中。
text = spark.read.text("./282.json")
rdd = text.rdd.map(lambda x: ProvDocument.deserialize(content=x))
print(rdd.take(1))
JSON文件很简单，是prov文件，在本地环境中它可以正常工作。这给了我以下错误：
21/12/10 11:33:16 ERROR Executor: Exception in task 0.0 in stage 58.0 (TID 56)
org.
浏览 3提问于2021-12-10得票数 0
回答已采纳
1回答
我正在了解火花，并想把一个列表(大约1000个条目)转换为一个火花df。
不幸的是，我得到了标题中提到的错误。我不知道是什么导致了这个错误，如果有人能帮我，我会非常感激的。到目前为止，这是我的代码：
# Pyspark SQL library
from pyspark.sql import SparkSession
from pyspark.sql.types import ArrayType
from pyspark.sql.types import StructField
from pyspark.sql.types import StructType
from pyspark.sql.t
浏览 7提问于2022-01-24得票数 0
回答已采纳
2回答
基于列值移除重复行
、、、、
我有一个PySpark RDD。我希望只在下一行中“列1”和“列2”匹配时消除重复的。数据是这样的：
2,10482422,0.18
2,10482422,0.4
2,10482423,0.15
2,10482423,0.43
2,10482424,0.18
2,10482424,0.49
2,10482425,0.21
2,10482425,0.52
2,10482426,0.27
2,10482426,0.64
2,10482427,0.73
在上面的示例中，我只希望通过删除其他重复行，为每一行设置一个第3列的最大值。
预期产出如下：
2,10482422,0.4
2,10482423,0
浏览 0提问于2018-05-05得票数 1
回答已采纳
1回答
我正在Python上开发一个Spark脚本(使用Pyspark)。我有一个函数，它用一些字段返回一个Row，包括
timestamp=datetime.strptime(processed_data[1], DATI_REGEX)
processed_data1是一个有效的日期时间字符串。
编辑显示完整的代码：
DATI_REGEX = "%Y-%m-%dT%H:%M:%S"
class UserActivity(object):
def __init__(self, user, rows):
self.user = int(user)
浏览 5提问于2017-01-19得票数 0
回答已采纳
我有以下代码:对于每个my_id，我尝试根据timestamp字段对amount字段进行排序：
output_rdd = my_df.rdd.map(lambda r: (r['my_id'], [r['timestamp'],[r['amount']]]))\
.reduceByKey(lambda a, b: sorted(a+b, key=(a+b)[0]))\
.map(lambda r: r[1])
但是，我得到了以下错误：
Py4JJava
浏览 14提问于2016-09-28得票数 0
回答已采纳
1回答
我是PySpark新手，为了提高我的PySpark技能，我买了一本书。我在使用函数的时候卡住了。
函数
def filterDuplicates (( userID，filterDuplicates) )：
(movie1, rating1) = ratings[0]
(movie2, rating2) = ratings[1]
return movie1 < movie2
由于两个连续的括号，我得到了错误。Step基本上得到了一个RDD，它基本上是一个复数列表，如下所示：
[(196, ((242, 3.0), (242, 3.0))), (196,
浏览 19提问于2019-08-04得票数 0
我有如下所示的元组列表
ls=[('c', 's'),('c', 'm'), ('c', 'p'), ('h', 'bi'), ('h', 'vi'), ('n', 'l'), ('n', 'nc')]
我想使用pyspark和groupByKey来制作：
nc=[['c','s', 'm', 'p'], [&
浏览 0提问于2018-04-18得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
