PySpark连接两个RDD会导致一个空RDD
的原因是两个RDD之间没有共享的键值对。在PySpark中，连接操作是通过键值对进行的，如果两个RDD之间没有共同的键值对，连接操作将返回一个空的RDD。
连接操作在数据处理和分析中非常常见，它可以将两个RDD基于键值对进行合并，从而实现数据的关联和整合。连接操作有多种类型，包括内连接、外连接和交叉连接等。
内连接（inner join）是连接操作中最常用的一种类型，它只返回两个RDD中键值对完全匹配的记录。具体而言，内连接会将两个RDD中具有相同键的键值对进行合并，并生成一个新的RDD。内连接可以用于数据的关联查询、数据过滤和数据整合等场景。
外连接（outer join）是连接操作中的另一种类型，它返回两个RDD中所有的键值对，如果某个RDD中的键值对在另一个RDD中没有匹配的记录，将用空值进行填充。外连接可以分为左外连接、右外连接和全外连接三种类型，具体的选择取决于需要保留哪些RDD中的记录。
交叉连接（cross join）是连接操作中的一种特殊类型，它将两个RDD中的所有记录进行组合，生成一个新的RDD。交叉连接会导致数据量的急剧增加，因此在实际应用中需要谨慎使用。
在PySpark中，可以使用join()方法来进行连接操作。例如，对于两个RDD rdd1 和 rdd2，可以使用以下代码进行内连接操作：
joined_rdd = rdd1.join(rdd2)对于外连接和交叉连接，可以使用leftOuterJoin()、rightOuterJoin()和cartesian()等方法来实现。
腾讯云提供了强大的云计算服务，包括云服务器、云数据库、云存储等产品，可以满足各种应用场景的需求。具体而言，腾讯云的云服务器（CVM）提供了高性能、可扩展的计算资源，可以用于部署和运行PySpark应用程序。腾讯云的云数据库（TencentDB）提供了可靠的数据存储和管理服务，可以用于存储和处理连接操作中的数据。腾讯云的云存储（COS）提供了安全、可靠的对象存储服务，可以用于存储和管理大规模的数据。
更多关于腾讯云产品的信息和介绍，请访问腾讯云官方网站：https://cloud.tencent.com/
相关·内容
1回答
加入PySpark不加入任何值
、、、、
在PySpark中，我想用键值对对两个RDD进行完全的外部连接，其中键可以是None。例如：
rdd1 = sc.parallelize([(None, "a"), (None, "b")])
rdd2 = sc.parallelize([(None, "c"), (None, "d")])
join_rdd = rdd1.join(rdd2)
看起来，PySpark加入了键为空的记录：
print(rdd1.join(rdd2).take(10))
>>> [(None, ('a', '
浏览 0提问于2016-03-02得票数 1
我在Databricks笔记本上运行一个作业，该笔记本连接到AWS RDS上的MySQL数据库并插入数据。当我手动运行笔记本时，我能够连接到端点URL并插入我的数据。现在，我的笔记本每30分钟运行一次。第一个作业成功，但之后的每个作业都失败，并显示以下错误： MySQLInterfaceError: MySQL server has gone away 然后，我再次尝试手动运行作业，但在tweets_pdf.to_sql(name='tweets', con=engine, if_exists = 'replace', index=False)上得到了相同的错
浏览 21提问于2020-07-10得票数 0
1回答
所有，我使用的是PySpark &需要连接两个RDD，但是要连接它们，我需要将每个RDD的所有元素按连接键分组，然后执行一个连接函数。这会造成额外的间接费用，我不知道周围的工作可以是什么。此外，这也造成了大量的任务，从而增加了要写入HDFS的文件数量，并大大减缓了整个进程--这里就是一个例子：
RDD1 = [join_col,{All_Elements of RDD1}] #derived by using groupby join_col)
RDD2 = [join_col,{All_Elements of RDD2}] #derived by using groupby join
浏览 6提问于2016-02-27得票数 0
回答已采纳
1回答
我有一个吡火花rdd，并试图使用一些自定义抽样比率将其转换为数据格式。但是我得到错误的，有时，空的rdd不能用来创建数据格式。
ValueError:无法减少()空RDD
下面是我的密码。就像我说过的，这并不是总是错误的。只有一些时候，它是失败的。
from pyspark import SparkContext
from pyspark.sql import SQLContext
sc = SparkContext()
sqlContext = SQLContext(sc)
myrdd = sc.parallelize([
(1, 638.55),
(2, 638.5
浏览 4提问于2019-11-21得票数 1
回答已采纳
我一直在试验PySpark RDD的分区和重新分区。
我注意到，当将一个小样本RDD从2个分区重新划分到6个分区时，只添加了几个空部分。
rdd = sc.parallelize([1,2,3,43,54,678], 2)
rdd.glom().collect()
>>> [[1, 2, 3], [43, 54, 678]]
rdd6 = rdd.repartition(6)
rdd6.glom().collect()
>>> [[], [1, 2, 3], [], [], [], [43, 54, 678]]
现在，我想知道这是否也发生在我的真实
浏览 2提问于2021-05-19得票数 1
回答已采纳
1回答
我正在尝试一些与pyspark相关的实验，在连接到AWS EMR实例的jupyter笔记本上。我有一个spark dataframe，它从s3读取数据，然后过滤掉一些东西。使用df1.printSchema()输出打印模式，如下所示： root
|-- idvalue: string (nullable = true)
|-- locationaccuracyhorizontal: float (nullable = true)
|-- hour: integer (nullable = true)
|-- day: integer (nullable = true)
|-- da
浏览 39提问于2019-05-09得票数 2
回答已采纳
当我通过rdd.repartition(1).saveAsTextFile(file_path)保存一对rdd时，会遇到一个错误。
Py4JJavaError: An error occurred while calling o142.saveAsTextFile.
: org.apache.spark.SparkException: Job aborted.
at org.apache.spark.internal.io.SparkHadoopWriter$.write(SparkHadoopWriter.scala:100)
at org.apache.spark.rdd.
浏览 1提问于2020-07-01得票数 0
回答已采纳
我正在尝试用MLLib在Pyspark中运行逻辑回归。模型运行了，但我无法获得任何指标。
我的数据是csv格式，我按如下方式进行转换：
def load(prefix):
lines = spark.read.text(prefix).rdd
parts = lines.map(lambda row: row.value.split(","))
ratingsRDD = parts.map(lambda p: Row(pct = str(p[0]), date = str(p[1]), res_burg_label=int(p[2]), com_burg
浏览 1提问于2018-03-17得票数 0
我正在尝试使用pyspark将两个没有共同之处(没有键是公共的)的csv文件组合成一个键值成对的rdd 让我们假设A.csv有 a
b
c 而B.csv有 1
2
3 在pyspark中有没有一个选项可以通过连接这两个来获得一个rdd，就像这样 a:1
b:2
c:3 当然，两个csv文件中的行数应该匹配。这是在pyspark中很容易做的事情，还是应该首先在常规的python中完成。也就是说，对这两个文件进行嵌套循环，然后创建一个元组元组，如((a,1)，(b,2)...)然后将其传递给parallelize。
浏览 13提问于2019-12-17得票数 0
在PySpark中，我有两个RDD，它们的结构是(key，list of list)：
input_rdd.take(2)
[(u'100',
[[u'36003165800', u'70309879', u'1']]),
(u'200',
[[u'5196352600', u'194837393', u'99']]) ]
output_rdd.take(2)
[(u'100',
[[u'875000', u
浏览 0提问于2017-10-12得票数 0
1回答
在下面的示例中，我误解了PySpark的性能。
我有几个DataFrame，因此我加入了它们。
print"users_data"
print users_data.show()
print"calc"
print calc.show()
print"users_cat_data"
print users_cat_data.show()
data1 = calc.join(users_data, ['category_pk','item_pk'], 'leftouter')
print "
浏览 0提问于2018-01-15得票数 2
回答已采纳
据我了解，在使用Spark接口时，当只需要一个或两个属性时，我们必须小心不要不必要地序列化一个完整的对象：()
当使用PySpark时，这是如何工作的？如果我有如下课程：
class C0(object):
def func0(arg):
...
def func1(rdd):
result = rdd.map(lambda x: self.func0(x))
这会导致对完整的C0实例进行筛选吗？如果是的话，有什么正确的方法来避免呢？
谢谢。
浏览 2提问于2016-04-08得票数 3
回答已采纳
1回答
Neo4j作为火花放电的数据源
、、、、
我有一个要求，我必须从Neo4j中提取数据，并从这些数据中创建Spark。我在我的项目中使用Python。连接器具有相同的用途，但它是用Scala编写的。所以我现在可以想出解决办法-
以小块/批的形式从neo4j查询数据，使用parallize()方法将每个块转换为Spark。最后，使用union()方法合并/合并所有的RDD，以获得单个RDD。然后我就可以对它们进行转换和操作。
另一种方法是从Neo4j读取数据，并从中创建一个卡夫卡制作者。然后使用Kafka作为星火的数据源。例如：
Neo4j -> Kafka ->火花
我想知道哪一个对大量数据更有效？如
浏览 7提问于2018-01-09得票数 2
我正在努力创建一个空的数据在火花(火花)。
我正在使用与这里讨论的类似的方法，，但它不起作用。
这是我的密码
df = sqlContext.createDataFrame(sc.emptyRDD(), schema)
这是错误
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/Users/Me/Desktop/spark-1.5.1-bin-hadoop2.6/python/pyspark/sql/context.py", line
浏览 4提问于2016-01-06得票数 34
回答已采纳
1回答
我试图从rdd中筛选值为"01-10-2019“的
print("\n ### Remove duplicates in merged RDD:")
insuredata = insuredatamerged_cache.distinct()
print("insuredata: ",type(insuredata))
print("\n ### Increase partition to 8 in merged RDD:")
insuredata.getNumPartitions()
insuredatarepart = insu
浏览 5提问于2020-02-08得票数 1
回答已采纳
我有一个带Anaconda的Dataproc集群。我已经创建了一个虚拟环境。在anaconda my-env中，因为我需要在那里安装开源RDkit，因此我再次安装了PySpark (而不是使用预先安装的)。现在，使用下面的代码，我在my-env中得到了错误，但在my-env之外没有得到错误 代码： from pyspark.sql.types import StructField, StructType, StringType, LongType
from pyspark.sql import SparkSession
from py4j.protocol import Py4JJavaEr
浏览 40提问于2020-01-29得票数 1
1回答
我有一个pyspark dataframe，并使用相同的dataframe来创建新的dataframe，并在最后连接它们。
例如：
source_dataframe = spark.createDataFrame(rdd, schema).cache()
df1 = function1(source_dataframe)
df2 = function2(source_dataframe)
df3 = function3(source_dataframe)
df4 = function4(source_dataframe)
每个函数都是相互独立的，并最终将它们连接起来以创建我的最终数据帧。
f
浏览 16提问于2020-05-27得票数 0
回答已采纳
1回答
正确使用大型广播变量的提示？
、、、、
我使用了一个大小约为100 MB的广播变量，我将其近似为：
>>> data = list(range(int(10*1e6)))
>>> import cPickle as pickle
>>> len(pickle.dumps(data))
98888896
运行在具有3个c3.2xLargeExecutor的集群上，并运行一个m3.large驱动程序，并使用以下命令启动交互式会话：
IPYTHON=1 pyspark --executor-memory 10G --driver-memory 5G --conf spark.drive
浏览 2提问于2016-05-26得票数 14
回答已采纳
我使用pyspark来运行这类连接：
rdd1=sc.textFile(hdfs_dir1).map(lambda row: (getKey1(row),getData1(row)))
rdd2=sc.textFile(hdfs_dir2).map(lambda row: (getKey2(row),getData2(row)))
result=rdd1.join(rdd2).collect()
作业执行第一个300任务的速度相当快(每个任务都有秒)，并且在到达任务301/308时挂起，即使我让它运行了几天。
我尝试使用不同的配置(工作人员数量、内存、cpus、内核、洗牌率)运行pyspark
浏览 0提问于2016-02-25得票数 4
回答已采纳
2回答
火花相交
、
我希望在pyspark中得到两个RDD的交集。它们看起来如下：
rdd1 = sc.parallelize(["abc","def", "ghi"])
rdd2 = sc.parallelize([["abc","123"],["df",345], ["ghi","678"])
是否可以使用pyspark的rdd操作符来获得：
intersection_rdd --> ["abc","123"] ["ghi",&#
浏览 4提问于2016-12-19得票数 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
回答已采纳
我正尝试在我的服务器上运行一个简单的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
1回答
我在努力学习阿帕奇火花。这是我正在尝试运行的代码。我正在使用pyspark api。
data = xrange(1, 10000)
xrangeRDD = sc.parallelize(data, 8)
def ten(value):
"""Return whether value is below ten.
Args:
value (int): A number.
Returns:
bool: Whether `value` is less than ten.
"""
if (value <
浏览 2提问于2015-06-09得票数 0
回答已采纳
我目前正在使用一个连接到RDS的EMR集群来收集两个表。
创建的两个RDD相当庞大，但我可以执行其他的.take(x)操作。
我还可以执行更复杂的操作，例如：
info_rdd = somerdd.map(lambda x: (x[1], x[2])).groupByKey().map(some_lambda)
apps_rdd = apps.join(info_rdd).map(lambda x: (x[0], (x[1][0], x[1][1][0], x[1][1][1])))
但是，执行以下操作来计算从RDS导入的不同用户的数量不起作用：
unique_users = rdd.dist
浏览 2提问于2017-02-14得票数 2
回答已采纳
我正在尝试运行pyspark分布式-kmodes示例：
import numpy as np
data = np.random.choice(["a", "b", "c"], (50000, 10))
data2 = np.random.choice(["e", "f", "g"], (50000, 10))
data = list(data) + list(data2)
from random import shuffle
shuffle(data)
# Create a Spark RDD
浏览 5提问于2017-05-09得票数 0
回答已采纳
我试图从原始的Dataframe "User_df“中获取每个列的列名和数据类型，到它们自己的列中，其中列的值分别是列名和数据类型。
我有88列和9500行。
当我尝试用一个列("ColumnDataTypes")创建一个Dataframe时，下面的代码是有效的，但当我试图为这两个列创建一个Dataframe时，代码就不起作用了：
col_types = User_df.dtypes
col_types2, col_types3 = [items[:] for items in zip(*col_types)]
rdd_col_names = sc.parallel
浏览 36提问于2022-08-16得票数 0
回答已采纳
1回答
你好，我试着用管道模型加载节省的管道。
selectedDf = reviews\
.select("reviewerID", "asin", "overall")
# Make pipeline to build recommendation
reviewerIndexer = StringIndexer(
inputCol="reviewerID",
outputCol="intReviewer"
)
produc
浏览 1提问于2018-07-10得票数 6
回答已采纳
1回答
我对pyspark是个新手。我一直在尝试将两个稀疏RDD相乘。whichI尝试生成两个稀疏矩阵的代码，我写了一个函数来乘以两个RDD，但我认为这不是解决方案，因为计算不是并行进行的。有人能帮我一下吗？如何并行地乘以RDD？我在这些网站上尝试了很多资源，但没有想出解决方案。
import findspark
findspark.init()
import numpy as np
import pyspark
import random
from scipy.sparse import rand
sc = pyspark.SparkContext(appName="matrix")
浏览 28提问于2019-11-02得票数 1
我在Google Cloud Dataproc上使用Spark streaming来执行一个框架(用Python编写)，它由几个连续的管道组成，每个管道代表Dataproc上的一个作业，它基本上是从Kafka队列读取并将转换后的输出写入Bigtable。所有管道组合在一起，每天通过2个群集处理数of的数据，一个具有3个工作节点，一个具有4个工作节点。
在Dataproc上运行这个Spark streaming框架在5月初(准确地说是5月3日)之前一直相当稳定:我们开始遇到频繁的套接字超时异常，这会终止我们的管道。它似乎与集群上的负载无关，因为它并没有显著增加。它在一天中也是随机发生的，我检查了
浏览 0提问于2016-05-24得票数 3
嗨，我在Notebook中有这段代码，并尝试编写python spark代码：
mydataNoSQL.createOrReplaceTempView("mytable")
spark.sql("SELECT * from mytable")
return mydataNoSQL
def getsameData(df,spark):
result = spark.sql("select * from mytable where temeperature is not null")
return result.rdd.sample(Fals
浏览 0提问于2017-06-04得票数 5
回答已采纳
我正试着在pycharm上运行pyspark。我已经连接了所有内容并设置了环境变量。我可以读取sc.textFile，但当我尝试从pyspark.sql读取csv文件时，出现了错误。
代码如下：
import os
import sys
from pyspark import SparkContext
from pyspark import SparkConf
from pyspark.sql import SQLContext
from pyspark.sql import SparkSession
# Path for spark source folder
os.environ[
浏览 5提问于2016-08-12得票数 2
回答已采纳
我正在做一个简短的应用程序，它根据漫威电影的出场次数告诉你，在漫威电影的大型数据集中，哪一个是最受欢迎的英雄。 我已经从Anaconda环境安装了Pyspark，也从控制台安装了Pyspark，试图解决这个错误，但没有结果。我还安装了用于conda的Java-jdk，但也没有结果。 我得到的错误如下： py4j.protocol.Py4JJavaError: An error occurred while calling o24.partitions.
: java.lang.IllegalArgumentException: java.net.URISyntaxException:
浏览 161提问于2019-03-21得票数 2
回答已采纳
1回答
我是Spark的新手。在将一个顶级外部函数传递到RDD_old.reduceByKey之后，当我试图从RDD_new收集结果时，出现了这个错误。
首先，我定义了一个treeStruct：
class treeStruct(object):
def __init__(self,node,edge):
self.node = nodeDictionary
self.edge = edgeDictionary
在那之后，我用sc.parallelize把两个treeStructs转换成了RDD：
RDD = sc.parallelize([treeStruct
浏览 0提问于2017-03-08得票数 0
我想使用Pyspark在数百个csv文件中读取，创建一个数据文件，即(粗略地)连接所有csv。因为每个csv都可以安装在内存中，但一次不超过一到两个，这似乎是一个很好的契合。我的策略不起作用，而且我认为这是因为我想在映射函数的内核函数中创建一个Pyspark dataframe，从而导致一个错误：
# initiate spark session and other variables
sc = SparkSession.builder.master("local").appName("Test").config(
"spark.driver.bindA
浏览 6提问于2022-10-03得票数 0
1回答
当我试图使用pyspark从S3获取数据时，我得到了一个空指针异常。我使用Hadoop2.4运行spark 1.6.1。我试着同时使用s3n和s3a。还尝试以下列方式设置配置：
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("fs.s3.impl", "org.apache.hadoop.fs.s3native.NativeS3FileSystem")
hadoopConf.set("fs.s3n.awsAccessKeyId", "aws-key")
浏览 3提问于2016-11-11得票数 2
回答已采纳
我正在尝试从S3中将大约一百万个压缩文本文件读入spark。每个文件的压缩大小在50 MB到80 MB之间。总共大约有6.5万亿字节的数据。
不幸的是，我遇到了一个内存不足的异常，我不知道如何解决。像这样简单的东西：
raw_file_list = subprocess.Popen("aws s3 ls --recursive s3://my-bucket/export/", shell=True, stdout=subprocess.PIPE).stdout.read().strip().split('\n')
cleaned_names = ["s3
浏览 20提问于2016-07-26得票数 4
回答已采纳
1回答
我需要用下面的方式用pyspark写一个if else条件。
schema = StructType([])
final = sqlCtx.createDataFrame(sc.emptyRDD(), schema)
if final.rdd.isEmpty:
print('abc')
else:
print('pqr')
但是，我似乎找不到正确的语法来检查dataframe是否为空。上面的final.rdd.isEmpty总是给出一个真值，即使最终的数据帧不是空的。
浏览 1提问于2018-10-11得票数 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
回答已采纳
1回答
在了解了Apache文档之后，我尝试使用mapPartition模块进行实验。在下面的代码中，我希望看到初始的RDD，就像在函数myfunc中一样，我只是在打印值之后返回迭代器。但是，当我在RDD上执行collect时，它是空的。
from pyspark import SparkConf
from pyspark import SparkContext
def myfunc(it):
print(it.next())
return it
def fun1(s
浏览 0提问于2017-03-23得票数 1
回答已采纳
如何获得一个小型Python脚本来连接到现有的Spark实例，并对现有的RDD进行操作？
我正处于Windows 10上使用Spark的早期阶段，在“本地”实例上尝试脚本。我正在使用最新的稳定构建的火花(星火2.0.1为Hadoop 2.7)。我已经为Hadoop2.7.3安装和设置了环境变量。我正在用Python来试验Pyspark和VisualStudio2015CommunityandPython。
我正在尝试构建一个大型引擎，我将在这个引擎上运行单独的脚本来加载、按摩、格式化和访问数据。我肯定有一种正常的方法，这不是火花的意义吗？
不管怎么说，这是我目前的经历。这通常是意料之中的。当我
浏览 2提问于2016-11-06得票数 5
我在用数据库。假设我有两个(我使用的是PySpark)：
df_source
df_target
如果df_source有以下模式：
root
|-- name: string (nullable = false)
|-- id: long (nullable = false)
|-- age: long (nullable = true)
df_target有以下模式：
root
|-- name: string (nullable = true)
|-- id: long (nullable = false)
|-- age: long (nullable =
浏览 1提问于2021-12-27得票数 0
回答已采纳
我有一个非常奇怪的错误火花数据，这导致一个字符串被计算为一个时间戳。
以下是我的设置代码：
from datetime import datetime
from pyspark.sql import Row
from pyspark.sql.types import StructType, StructField, StringType, TimestampType
new_schema = StructType([StructField("item_id", StringType(), True),
StructField(
浏览 2提问于2017-02-03得票数 8
我在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
总体问题:从csv文件生成模式并将其应用于数据文件。我有一个只有一列的RDD，我想用它做一个字符串。因此，我正在使用下面的代码来实现这一点，它在中工作得很好，但是在火花作业中失败。
schema = metadata.map(lambda l: l).reduce(lambda l, m: l+ "," + m)
所以输出应该是'id，name，age‘。但是，当我执行任务时，我会得到错误：
Exception: It appears that you are attempting to broadcast an RDD or reference
an RDD from
浏览 5提问于2016-03-17得票数 0
回答已采纳
所以我有两个RDD1 (假设是RDD1和RDD2)，每个都有一个数字列表。这两个列表的大小相同。我想创建一个RDD3，其中RDD3中的每个元素都是RDD1和RDD2的相应元素的相加。如何在python中使用pyspark函数完成此操作？
浏览 12提问于2020-07-10得票数 0
局部模式运行
计算机内存：16‘s
计算机可用磁盘空间：500‘s
输入文件大小：50
错误消息：java.lang.OutOfMemoryError: Java堆空间
代码:在命令窗口中使用.\pyspark后，输入到windows命令提示符中的
码
from pyspark import SparkConf, SparkContext
import pyspark
rdd = sc.textFile('file:///myBigFile.xml')
rdd.persist(pyspark.StorageLevel.DISK_ONLY)
r
浏览 2提问于2019-01-31得票数 0
在检查了所有找到的源之后，datastax-spark-cassandra连接器支持在cassandra中使用scala和java中的rdd自动创建表。特别是对于pyspark，有另一个包可以完成这项工作-- 。但是即使有了这个包也不能自动创建表格。对于dataframe，我根本找不到任何选项。我刚接触pyspark和cassandra，任何帮助都是非常感谢的。尝试仅使用anguenot包作为依赖项。Spark版本: 2.4.7 Cassandra :最新docker镜像
Pyspark shell >> pyspark --packages anguenot/pyspark-ca
浏览 8提问于2021-03-03得票数 2
我正在使用Spark3.1.1版本的PySpark DataFrame API进行本地设置。在读取数据、执行一些转换等之后，我将DataFrame保存到磁盘。输出目录与part-0000*文件一起创建，输出目录中也有_SUCCESS文件。但是，我的part-0000*总是空的，即零字节。
我试着用parquet和csv格式编写它，结果是一样的。在编写之前，我调用了df.show()以确保DataFrame中有数据。
### code.py ###
from pyspark.sql import SparkSession
from pyspark.sql import functions as
浏览 8提问于2022-05-11得票数 1
我有一个Kafka流作业(Spark version2.4.5)在kubernetes上运行，有一个驱动程序和4个executors.The，该作业的目的是消费来自PySpark主题的数据，并对它们进行处理，然后将最终数据加载到ElasticSearch。
问题：
有时，我们会遇到Kafka集群的问题。因此，PySpark流作业将不会获得任何要处理/消费的数据，即acceptable.However，即使在Kafka群集变得正常之后，PySpark流作业也不会消耗/处理来自Kafka topic.The作业的任何进一步记录
错误:在驱动程序或任何执行器上找不到错误。
其中一个executors
浏览 15提问于2021-11-25得票数 1
1回答
我在pyspark有两个rdd
rdd1=sc.parallelize(['a','b'])
rdd2=sc.parallelize(['c','d'])
我想生成一个rdd，其中包含每个rdd的一个元素。(a，c)，(b，c)，(a，d)，(b，d)我试过了
rdd3=rdd1.map(lambda x:x)+rdd2.map(lambda y:y)
它失败了
浏览 13提问于2019-11-17得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
