Pyspark:如何计算RDD中每个等距区间的个数
Pyspark是一个用于大规模数据处理的Python库，它提供了丰富的功能和工具来处理和分析大数据集。在Pyspark中，RDD（弹性分布式数据集）是一种基本的数据结构，它可以在集群上并行处理和操作。
要计算RDD中每个等距区间的个数，可以按照以下步骤进行：
- 首先，确定等距区间的范围和间隔。例如，如果要计算0到100之间每个10个数的区间个数，范围是0到100，间隔是10。
- 使用Pyspark的RDD功能，将数据集加载到RDD中。可以使用
sc.parallelize()方法将数据集转换为RDD。 - 使用
map()函数将RDD中的每个元素映射到对应的区间。在映射函数中，可以使用整除（//）操作符将元素除以间隔，然后取整得到区间的索引。 - 使用
countByValue()函数统计每个区间的个数。这将返回一个包含区间索引和对应个数的字典。
下面是一个示例代码：
from pyspark import SparkContext
# 创建SparkContext对象
sc = SparkContext("local", "Pyspark Example")
# 加载数据集到RDD
data = sc.parallelize([1, 5, 12, 18, 25, 30, 35, 40, 45, 50, 55, 60, 65, 70, 75, 80, 85, 90, 95, 100])
# 定义等距区间范围和间隔
start = 0
end = 100
interval = 10
# 将每个元素映射到对应的区间
mapped_data = data.map(lambda x: (x // interval, x))
# 统计每个区间的个数
count_by_interval = mapped_data.countByValue()
# 打印结果
for interval, count in count_by_interval.items():
print("区间 {}-{} 的个数为: {}".format(interval * interval, (interval + 1) * interval, count))这段代码将输出每个等距区间的个数，例如：
区间 0-10 的个数为: 2
区间 10-20 的个数为: 2
区间 20-30 的个数为: 2
区间 30-40 的个数为: 2
区间 40-50 的个数为: 2
区间 50-60 的个数为: 2
区间 60-70 的个数为: 2
区间 70-80 的个数为: 2
区间 80-90 的个数为: 2
区间 90-100 的个数为: 2在腾讯云的产品中，与Pyspark相对应的是腾讯云的大数据计算服务TencentDB for Apache Spark，它提供了强大的大数据处理和分析能力，可以帮助用户高效地处理和分析大规模数据集。您可以通过访问以下链接了解更多关于TencentDB for Apache Spark的信息：TencentDB for Apache Spark。
相关·内容
1回答
沿阿基米德螺旋等距放置点
、、、
我有一个阿基米德螺旋，由参数方程x = r t * cos(t)和y = r t * sin(t)决定。
我需要沿着螺旋线等距离放置n点。等距的确切定义并不重要--它只需是近似的。
仅以r、t和n作为参数，如何计算每个等距点的坐标？
浏览 1提问于2017-06-24得票数 3
回答已采纳
我正在使用一些dataframe df创建一个数据样本
rdd = df.limit(10000).rdd
这个操作需要相当长的时间(为什么呢？它不能在10000行之后省略吗？)，所以我假设我现在有了一个新的RDD。
但是，当我现在使用rdd时，每次访问它都是不同的行。就像它再次重新采样一样。缓存RDD有一点帮助，但这肯定不是保存？
背后的原因是什么？
更新:这里是Spark 1.5.2的复制品
from operator import add
from pyspark.sql import Row
rdd=sc.parallelize([Row(i=i) for i in range(100
浏览 2提问于2016-05-11得票数 15
有一个火花放电代码段。
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
回答已采纳
我有一个RDD[Double]，我想把RDD分成k等距离间隔，然后计算每个等距离间隔的个数。 例如，RDD类似于[0,1,2,3,4,5,6,6,7,7,10]。我想把它分成相等的间隔，所以间隔是[0,1), [1,2), [2,3), [3,4), [4,5), [5,6), [6,7), [7,8), [8,9), [9,10]。 正如您所看到的，RDD的每个元素都将处于其中一个间隔中。然后我想计算每个区间的个数。这里，[0,1),[1,2),[2,3),[3,4),[4,5),[5,6)中只有一个元素，而[6,7)和[7,8)都有两个元素。[9,10]有一个元素。 最后，我期望得到一个
浏览 21提问于2020-07-01得票数 0
回答已采纳
我正在使用MLlib和Python(Pyspark)，我想知道在执行我的代码之前在内存中创建的RDD的数量。我正在RDD上执行事务和操作，所以我想知道在内存中创建的RDD的总数。
浏览 2提问于2019-12-09得票数 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
所以我有两个RDD1 (假设是RDD1和RDD2)，每个都有一个数字列表。这两个列表的大小相同。我想创建一个RDD3，其中RDD3中的每个元素都是RDD1和RDD2的相应元素的相加。如何在python中使用pyspark函数完成此操作？
浏览 12提问于2020-07-10得票数 0
1回答
我正在使用pyspark来分析一个数据集，对于为什么下面的代码正确工作，我感到有点惊讶，尽管我使用的变量是而不是广播。
所讨论的变量是video，它用于函数filter之后的联接。
seed = random.randint(0,999)
# df is a dataframe
# video is just one randomly sampled element
video = df.sample(False,0.001,seed).head()
# just a python list
otherVideos = [ (22,0.32),(213,0.43) ]
# transfo
浏览 4提问于2015-10-26得票数 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
我有一个SQLContext数据框架，它是由几个数字列组成的熊猫数据框架。我想使用pyspark.mllib.stats包进行多元统计分析。统计函数期望向量的RDD。我无法将此数据帧转换为向量的RDD。有办法转换数据帧吗？
代码：
rdd = sqlCtx.createDataFrame(df_new)
summary = Statistics.colStats(rdd)
我得到了df_new
df_new = df.applymap(lambda s: dic.get(s) if s in dic else s) #df is a pandas dataframe
我在总结行中得到一个
浏览 0提问于2015-07-01得票数 1
回答已采纳
下面的代码片段试图将一个简单的函数应用于一个PySpark RDD对象：
import pyspark
conf = pyspark.SparkConf()
conf.set('spark.dynamicAllocation.minExecutors', 5)
sc = SparkContext(appName="tmp", conf=conf)
sc.setLogLevel('WARN')
fn = 'my_csv_file'
rdd = sc.textFile(fn)
rdd = rdd.map(lambda line: l
浏览 3提问于2016-03-20得票数 1
回答已采纳
2回答
我们都知道斯派克在内存中做计算。我只是对下面的内容感到好奇。
如果我从HDFS中在我的RDD外壳中创建了10个pySpark，这是否意味着所有这10个RDD的数据都将驻留在Spark内存中？
如果我不删除RDD，它会永远存在于记忆中吗？
如果我的数据集(文件)大小超过可用RAM大小，数据将存储在哪里？
浏览 5提问于2016-11-22得票数 21
回答已采纳
1回答
我有一个pyspark RDD，它有大约200万个元素。我不能一次收集它们，因为它会导致OutOfMemoryError异常。 如何批量采集？ 这是一个潜在的解决方案，但我怀疑还有更好的方法:收集一个批处理(使用take、https://spark.apache.org/docs/3.1.2/api/python/reference/api/pyspark.RDD.take.html#pyspark.RDD.take)，然后从该批处理中的RDD中删除所有元素(使用filter、https://spark.apache.org/docs/3.1.2/api/python/reference/a
浏览 29提问于2021-10-12得票数 0
回答已采纳
我有一个PySpark RDD，其中每一行都是Pandas对象，我需要获得一个数据have结果。
这类似于已经回答的以下问题，但所接受的解决方案使用的是仅在旧版本(3.0.0)中可用的PySpark功能。
调用数据帧转换
RDD.toDF()
直接导致错误。数据的真值是模糊的。
这与收集RDD的所有行一起操作。
rdd.collect()
但这是非常缓慢的，并将所有数据返回给一个驱动程序。
调用RDD中每个数据的火花数据格式转换
.toDF()
也不能工作，因为双和字符串之间存在marge类型错误，所以我必须将Pandas数据格式与
.astype(str)
这很慢。
有什么更好的
浏览 6提问于2022-08-04得票数 0
4回答
PySpark计算相关性
、、、、
我想使用pyspark.mllib.stat.Statistics.corr函数来计算pyspark.sql.dataframe.DataFrame对象的两列之间的相关性。corr函数期望接受rdd of Vectors对象。如何将df['some_name']列转换为rdd of Vectors.dense对象？
浏览 12提问于2016-06-03得票数 16
回答已采纳
1回答
如您所知，我对使用Pyspark相当陌生，我的RDD设置如下：(ID、名字、姓氏、地址) (ID、名字、姓氏、地址)--无论如何，我可以数一数我在RDD中存储的这些记录中的多少，例如RDD中的所有ID。这样输出结果就会告诉我我有5个。我尝试过使用RDD.count()，但这似乎只是返回了我的数据集中共有多少项。
浏览 0提问于2018-11-25得票数 0
3回答
我试图创建一个数据，使用随机均匀分布在星火。我找不到关于如何创建数据文件的任何内容，但是当我阅读文档时，我发现pyspark.mllib.random有一个RandomRDDs对象，它有一个uniformRDD方法，它可以从随机的均匀分布中创建rdds。
但问题是它不能产生二维的rdd。是否有一种方法可以创建一个二维的rdd或者(最好是)数据？
我可以创建一些rdd，并使用它们来创建一个数据格式，但是我使用的数据集有许多字段(100+)，并且创建了100多个rdd，然后压缩它们似乎没有效率。
浏览 1提问于2018-05-16得票数 7
回答已采纳
1回答
我正在尝试查看在partitionBy保存后续操作之后，rdd上的persist()是否会保存后续操作，而spark ui似乎表明我没有保存任何操作。 如果persist正常工作，我认为阶段7或阶段8应该被跳过 (无论哪种情况，我的测试代码都可能是错误的，请让我知道。) ? 下面是我使用的代码 from pyspark import SparkContext, SparkConf
from pyspark.rdd import portable_hash
from pyspark.sql import SparkSession, Row
from pyspark.storag
浏览 35提问于2019-06-22得票数 1
我想把我的results1 numpy数组转换成一个数据帧。根据记录，results1看起来像 array([(1.0, 0.1738578587770462), (1.0, 0.33307021689414978),
(1.0, 0.21377330869436264), (1.0, 0.443511435389518738),
(1.0, 0.3278091162443161), (1.0, 0.041347454154491425)]). 我想将上面的代码转换为pyspark RDD，其中包含标记为"limit“(元组中的第一个值)和"pro
浏览 45提问于2020-10-28得票数 0
我使用的是亚马逊EC2，我的主服务器和开发服务器是一体的。我有另一个单个worker的实例。
我是个新手，但我已经设法让spark在独立模式下工作了。现在我正在尝试集群。主节点和辅助节点处于活动状态(我可以看到它们的webUI，它们正在工作)。
我有Spark 2.0，并且我已经安装了最新的Anaconda 4.1.1，它与Python 3.5.2一起提供。在worker和master中，如果我进入os.version_info并执行PYSPARK_PYTHON，我会得到3.5.2，我也正确地设置了所有的环境变量(如stackoverflow和google上的其他帖子)。
无论如何，在任何地方
浏览 1提问于2016-08-14得票数 1
由于我是星火社区的新手，任何人都能解释如何从数据库表创建PySpark RDD吗？我可以使用PySpark方法的textFile()方法从CSV文件创建SparkContext RDD。但我不知道从数据库表中创建PySpark RDD。
浏览 0提问于2018-02-10得票数 1
回答已采纳
我想要执行"execute“方法，所以为了避免对Spark的惰性计算，我想做一个动作(saveAsTextFile)，如代码所示：
def execute(line1):
line = line1.split(',')
print('Hi')
session = driver.session()
#vérifions si les noeuds n'existent pas encore et si oui créons les
session.run("MERGE (n:Person {T
浏览 1提问于2018-08-09得票数 1
回答已采纳
在客户端模式下启动火花放电。bin/pyspark --master yarn-client --num-executors 60，shell上的导入numpy很好，但是在kmeans中失败了。不知怎么的，执行者们没有安装过，这是我的感觉。我在任何地方都找不到好的解决办法，让工人们知道“无生气”的事。我试着设置PYSPARK_PYTHON，但这也不起作用。
import numpy
features = numpy.load(open("combined_features.npz"))
features = features['arr_0']
features.
浏览 7提问于2016-02-05得票数 19
回答已采纳
我有一个PySpark RDD groupby转换，它需要混洗到10000个分区，以减少在每个分区上操作所需的内存。但在此之后，我希望将输出写入较少数量的文件，如100个。如果我在输出上使用coalesce，它会自动将shuffle操作更改为100，这会很慢，并导致内存错误。我尝试过的另一种方法是在groupby之后持久化rdd (使用count()强制物化)，然后使用coalesce。这种方式实现了我想要的结果，但它增加了将RDD序列化到磁盘上的检查点的开销。我也尝试过使用localCheckpoint()，据我所知，它无论如何都需要缓存/持久化rdd。所以我想知道有没有其他更好的方法？换句
浏览 0提问于2020-07-07得票数 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
回答已采纳
2回答
这里我使用pyspark做了一个简单的字数统计，我使用了cache方法，但似乎没有在第二次运行代码时加速：
from pyspark import SparkContext
from time import time
if __name__=='__main__':
sc=SparkContext()
rdd=sc.textFile("shakespear.txt")
rdd.cache()
t1=time()
rdd.flatMap(lambda r:r.split(" "))\
.map(
浏览 0提问于2017-07-23得票数 0
1回答
我是Apache的新手，我有几个基本的问题，在阅读火花材料时，我无法理解。每一种材料都有自己的解释风格。我正在使用PySpark朱庇特笔记本在Ubuntu上练习。
根据我的理解，当我运行以下命令时，testfile.csv中的数据被分区并存储在相应节点的内存中。
rdd1 = sc.textFile("testfile.csv")
我的问题是，当我运行以下转换和操作命令时，rdd2数据将存储在哪里。
1.它是否存储在记忆中？
rdd2 = rdd1.map( lambda x: x.split(",") )
rdd2.count()
我知道rdd2中的数据在关
浏览 2提问于2016-07-26得票数 6
回答已采纳
2回答
我有很大的时间序列数据，数据格式是：(arrival_time，键，值)，时间单位是秒，例如：
0.01, k, v
0.03, k, v
....
1.00, k, v
1.10, k, v
1.20, k, v
1.99, k, v
2.00, k, v
...
我需要做的是获取整个数据每秒的行数。到目前为止，我使用的是pySpark，我的代码如下：
linePerSec = []
lo = rdd.take(1)[0]
hi = lo + 1.0
end = rdd.collect()[-1][0]
while(hi < end):
number = rdd.filter
浏览 1提问于2015-11-16得票数 4
回答已采纳
我想使用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
在PySpark RDD中，如何使用foreachPartition()打印出每个分区的第一条记录？
浏览 0提问于2017-09-04得票数 2
我正在查看RDD的分区
>>> rdd = sc.parallelize(range(10), 5)
>>> rdd.collect()
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
>>> rdd.glom().collect()
[[0, 1], [2, 3], [4, 5], [6, 7], [8, 9]]
>>> rdd.mapPartitions(lambda x: [x]).collect()
[range(0, 2), range(2, 4), range(4, 6), range(6
浏览 2提问于2018-06-28得票数 1
我正在尝试运行中给出的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
回答已采纳
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
回答已采纳
0回答
当在另一个数据帧上执行UDF时，如何引用一个pyspark数据帧？
这里有一个虚拟的例子。我正在创建两个数据帧scores和lastnames，每个数据帧中都有一个列，该列在两个数据帧中是相同的。在scores上应用的UDF中，我希望过滤lastnames并返回在lastname中找到的字符串。
from pyspark import SparkContext
from pyspark import SparkConf
from pyspark.sql import SQLContext
from pyspark.sql.types import *
sc = SparkContext(&#
浏览 6提问于2016-12-30得票数 6
回答已采纳
1回答
我正在多次运行一个涉及spark并行化的程序。这个程序在最初的几次迭代中运行良好，但由于内存问题而崩溃。我使用的是Spark 2.2.0和Python2.7，我在亚马逊EC2上运行我的测试，内存为30g。
下面是我的Spark设置：
conf = pyspark.SparkConf()
conf.set("spark.executor.memory", '4g')
conf.set('spark.executor.cores', '16')
conf.set('spark.cores.max', '16&#
浏览 0提问于2017-10-12得票数 1
我有一个代码(kafka_producer.py)，从csv >>中读取，创建Pandas dataframe >>，将熊猫的数据转换为星星之火>>调用>>方法，将消息发送给kafka。df.foreachPartition(self.send_to_kafka)正在抛出PicklingError: Could not serialize object: TypeError: can't pickle _thread.RLock objects。
代码如下：
def get_kafka_producer(topic):
kafka
浏览 37提问于2022-05-27得票数 0
回答已采纳
假设我在pyspark工作。我有一个RDD，rdd = sc.Textfile(file.txt)，我调用rdd.persist()，然后重新分配它：rdd = rdd.filter(lambda x: 'hi' in x)。
现在是将sc.Textfile(file.txt)持久化在内存中，还是将rdd.filter(lambda x: 'hi' in x)持久化？
浏览 0提问于2018-02-20得票数 0
我用火花放电来处理数据。数据如下：
8611060350280948828b33be803 4363 2017-10-01
8611060350280948828b33be803 4363 2017-10-02
4e5556e536714363b195eb8f88becbf8 365 2017-10-01
4e5556e536714363b195eb8f88becbf8 365 2017-10-02
4e5556e536714363b195eb8f88becbf8 365 2017-10-03
4e5556e536714363b195eb8f88becbf8
浏览 4提问于2017-12-01得票数 0
我一直试图阅读我的数据从一个卡夫卡的主题，并将它写到一个拼花文件。到目前为止，除了foreachRdd函数之外，一切都在工作。当我在dstream上使用映射时，我可以看到数据，但是在使用foreachRdd的下一步，Rdd总是空的，我不知道为什么。
我的环境是Ubuntu运行卡夫卡和火花独立。我用的是火花放电壳。我对python还不熟悉，所以在语法方面我仍然遇到了很多问题，我不知道这是否是我的问题所在。
任何帮助或洞察力都将不胜感激。
下面是我的代码的一个副本，我一直把它粘贴在pyspark外壳中
from pyspark import SparkContext
from pyspark.st
浏览 3提问于2018-01-07得票数 1
回答已采纳
我正在查询一个CosmosDb集合，并且能够打印结果。当我试图将结果存储到一个火花DataFrame时，它就失败了。
以该网站为例：
遵循上述链接中的准确步骤。另外，尝试下面的方法
df = spark.createDataFrame(dataset)
这会引发此错误：
ValueError:某些类型在推断后无法确定
ValueError回溯(最近一次调用)
在()
25打印(数据集)
26
-> 27 df =spark.createDataFrame(数据集)
28 df.show()
29 /databricks/spark/python/pyspark/sql/sessio
浏览 2提问于2019-05-01得票数 1
1回答
文件videos.csv如下所示
2482cal-2792-48da,Action,Comedy
099acca-8888-48ca,Action,Comedy
在Scala中，Scala可以使用.drop(1)删除RDD列1，以便对所有行删除rdd列1，例如482cal-2792-48da和099acca-8888-48ca。
我想知道如何在PySpark rdd中PySpark中这样做，如下所示，在第2行代码中
火花Scala
val records = sc.textFile("file:///home/videos.csv")
val words = records.
浏览 2提问于2016-12-29得票数 3
1回答
我正在使用星火流创建一个系统来丰富来自cloudant数据库的输入数据。例子-
Incoming Message: {"id" : 123}
Outgoing Message: {"id" : 123, "data": "xxxxxxxxxxxxxxxxxxx"}
驱动程序类的代码如下：
from Sample.Job import EnrichmentJob
from Sample.Job import FunctionJob
import pyspark
from pyspark.streaming.kafka import
浏览 6提问于2016-12-30得票数 4
回答已采纳
1回答
如何从PySpark中检索RDD的依赖关系？
Scala有方法：
public final scala.collection.Seq<Dependency<?>> dependencies()
我在PySpark 中看不到等效的方法。有没有其他方法可以在PySpark中检索RDD依赖关系？
我见过toDebugString方法。但这真的只适用于交互式spark会话。
浏览 2提问于2017-12-25得票数 0
1回答
我试图用CDH在纱线上运行电火花，在这个生火的主程序中，有一条类似于rdd.pipe("XXX.sh")的语句，每次我运行它时，都会弹出一个被拒绝的错误，我该怎么办来解决这个错误呢？谢谢。
错误日志信息是blow：
文件"/data/yarn/nm/usercache/work/appcache/application_1495632173402_0079/container_1495632173402_0079_01_000001/pyspark.zip/pyspark/rdd.py"，第2346行，pipeline_func文件"/data/yarn
浏览 3提问于2017-05-25得票数 0
2回答
我的火花版本是1.3，我使用的是电火花。
我有一个叫做df的大数据。
from pyspark import SQLContext
sqlContext = SQLContext(sc)
df = sqlContext.parquetFile("events.parquet")
然后选择dataframe的几个列，并尝试计算行数。这个很好用。
df3 = df.select("start", "end", "mrt")
print(type(df3))
print(df3.count())
然后我应用用户定义的函数将一个列从一个字
浏览 2提问于2015-10-26得票数 4
回答已采纳
我在Google Cloud Dataproc上使用Spark streaming来执行一个框架(用Python编写)，它由几个连续的管道组成，每个管道代表Dataproc上的一个作业，它基本上是从Kafka队列读取并将转换后的输出写入Bigtable。所有管道组合在一起，每天通过2个群集处理数of的数据，一个具有3个工作节点，一个具有4个工作节点。
在Dataproc上运行这个Spark streaming框架在5月初(准确地说是5月3日)之前一直相当稳定:我们开始遇到频繁的套接字超时异常，这会终止我们的管道。它似乎与集群上的负载无关，因为它并没有显著增加。它在一天中也是随机发生的，我检查了
浏览 0提问于2016-05-24得票数 3
到目前为止，Spark还没有为流数据创建DataFrame，但是当我进行异常检测时，使用DataFrame进行数据分析更方便、更快。我已经做了这一部分，但是当我尝试用流式数据进行实时异常检测时，问题就出现了。我尝试了几种方法，但仍然无法将DStream转换为DataFrame，也无法将DStream中的RDD转换为DataFrame。
下面是我最新版本的代码的一部分：
import sys
import re
from pyspark import SparkContext
from pyspark.sql.context import SQLContext
from pyspark.sql
浏览 2提问于2016-02-06得票数 10
回答已采纳
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
回答已采纳
我正在使用Geoip2的python库和pySpark来获取一些I的地理地址。我的代码是：
geoDBpath = 'somePath/geoDB/GeoLite2-City.mmdb'
geoPath = os.path.join(geoDBpath)
sc.addFile(geoPath)
reader = geoip2.database.Reader(SparkFiles.get(geoPath))
def ip2city(ip):
try:
city = reader.city(ip).city.name
except:
浏览 6提问于2015-11-16得票数 8
回答已采纳
如果我已经存在数据帧，且我想要向数据框中添加新列
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
from pyspark.sql import Row
numbers=[1,2,30,4]
rdd1 = sc.parallelize(li)
row_rdd = rdd1.map(lambda x: Row(x))
test_df = sqlContext.createDataFrame(row_rdd,['numbers'])
------------------------------------
浏览 0提问于2020-10-05得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
