这是实际的管道。我正在将文本加载到RDD中。然后我把它清理干净。= rdd1.flatMap(Func)
rdd3 = rdd2.filter(lambda x: x notin stopwords) # filter out stopwords
Out:['a',
浏览 21提问于2019-04-27得票数 0
所以我有两个RDD1 (假设是RDD1和RDD2)，每个都有一个数字列表。这两个列表的大小相同。我想创建一个RDD3，其中RDD3中的每个元素都是RDD1和RDD2的相应元素的相加。如何在python中使用pyspark函数完成此操作？
浏览 12提问于2020-07-10得票数 0
1回答
我有一个pyspark RDD，它有大约200万个元素。我不能一次收集它们，因为它会导致OutOfMemoryError异常。 如何批量采集？这是一个潜在的解决方案，但我怀疑还有更好的方法:收集一个批处理(使用take、https://spark.apache.org/docs/3.1.2/api/python/reference/api/pyspark.RDD.take.html
浏览 29提问于2021-10-12得票数 0
回答已采纳
我正在尝试使用pyspark将两个没有共同之处(没有键是公共的)的csv文件组合成一个键值成对的rdd 让我们假设A.csv有 ac 而B.csv有 13 在pyspark中有没有一个选项可以通过连接这两个来获得一个rdd，就像这样 a:1c:3 当然，两个csv文件中的行数应该匹配。这是在pyspark中很容易做的事情，还是应该首先在常规<
浏览 13提问于2019-12-17得票数 0
我有一个SparkJob，它从在N项之间创建一个成对的分数矩阵开始。虽然密集，这是相当快-到大约20K元素，之后，它似乎被困了很长时间。我在多次尝试中看到的最后一个日志行是“清除累加器”，我将下面的代码块附加到下面，以便用随机创建的50K元素数据集来重现这个问题。笛卡尔产品的速度相当快，结果的RDD计数会在几分钟内(25亿行)返回，但是第二次计数会停留两个多小时，日志或Spark中
浏览 2提问于2016-06-29得票数 1
我是Spark和Scala的新手。我对reduceByKey函数在Spark中的工作方式感到困惑。pairs = lines.map(s => (s, 1))映射函数很清晰:s是键，它指向data.txt中的行但是，我不知道reduceByKey在内部是如何工作的？"a“是否指向关键字？或者，"a“是否指向"s"？那么a+ b代表什么呢？它
浏览 42提问于2015-05-10得票数 66
回答已采纳
1回答
星图变换
、、
为什么Map转换将单行作为单个元素？是否有任何方法来改变这一点，以便它将匹配一个模式，而不是默认的单行？示例：
映射是Apache中的一个转换操作，它应用于RDD的每个元素，并将结果作为新的RDD返回。
浏览 5提问于2017-05-20得票数 2
回答已采纳
由于我是星火社区的新手，任何人都能解释如何从数据库表创建PySpark RDD吗？我可以使用PySpark方法的textFile()方法从CSV文件创建SparkContext RDD。但我不知道从数据库表中创建PySpark RDD。
浏览 0提问于2018-02-10得票数 1
回答已采纳
我目前正在自学Spark programming，并试图用PySpark重新编写一个现有的Python应用程序。然而，我仍然对如何在PySpark中使用常规Python对象感到困惑。我了解Spark中的分布式数据结构，如RDD、DataFrame、Datasets、vector等。Spark有自己的转换操作和动作操作，如.map()、.reduceByKey()来操作这些对象。但是，如果我在PySpark中<
浏览 34提问于2017-03-01得票数 1
回答已采纳
1回答
Neo4j作为火花放电的数据源
、、、、
我有一个要求，我必须从Neo4j中提取数据，并从这些数据中创建Spark。我在我的项目中使用Python。连接器具有相同的用途，但它是用Scala编写的。所以我现在可以想出解决办法-
以小块/批的形式从neo4j查询数据，使用parallize()方法将每个块转换为Spark。最后，使用union()方法合并/合并所有的RDD，以获得单个RDD。如果有更好的
浏览 7提问于2018-01-09得票数 2
我刚开始接触spark，并试图理解普通RDD和配对RDD之间的区别。使用成对RDD而不是普通RDD的用例有哪些？如果可能，我想通过一个例子来了解pair RDD的内部结构。谢谢
浏览 1提问于2016-05-06得票数 15
我有这样的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
回答已采纳
我需要使用Pyspark从Elasticsearch读取数据。我正在尝试在pyspark中设计一个流，如下所示-ii) foreach -in-rdd1"org.elasticsearch.hadoop.mr.LinkedMapWritable"，rdd2 = sc.newAPIHadoopRDD("org.
浏览 0提问于2016-07-11得票数 0
当我试图将一个函数传递给Spark的map方法时，我遇到了一些问题。我的问题似乎是在功能，但不确定它。我的功能是这样的： rowDict = row.asDict() rowDicthash_column返回一个Row()对象，或者通过将函数简化为只返回("Hello")，但仍然收到相同的错误。/sql/session.py"
浏览 34提问于2022-10-28得票数 0
我正在使用cassandra 2.0.3，我想使用pyspark (Apache Spark Python API)从cassandra数据创建一个RDD对象。请注意:我不想做导入CQL，然后从pyspark API查询CQL，而是我想创建一个RDD，我想在上面做一些转换。
我知道在Scala中可以做到这一点，但我不知道如何在pyspark中做到这一点。
浏览 10提问于2013-12-30得票数 9
我正在使用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
回答已采纳
1回答
我是spark-scala开发的新手。我试图使用scala在spark中创建一个映射值，但得到了类型不匹配错误。
浏览 1提问于2015-11-08得票数 0
我从Hbase加载了数据，并对这些数据进行了一些操作，然后创建了一个成对的RDD。我想在我的下一个函数中使用这个RDD的数据。我在RDD中有50万条记录。你能建议从成对的RDD中按键读取数据的性能有效的方法吗？
浏览 45提问于2016-08-01得票数 0
我有一个RDD，每一项都是表格。每个项目都是一个键值对，该值是一个元素列表。我想解压列表，这样我就可以创建一个新的RDD，每个项目都包含一个元素，如下所示：(key, ele2)..(key, ele4)我试着做
RDD.flatmap(lamb
浏览 2提问于2015-12-13得票数 3
回答已采纳
我想(在本地)测试将一些值插入到mongo数据库中。table.find_one({})
import pysparkimport pymongo servers=(("mong
浏览 2提问于2019-08-20得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
