我编写了一个Pyspark程序，该程序获取同一输入文件的两个完全相同的副本，并将数据转换为两个新文件，每个文件都有自己的格式。我将这两个文件读入dataframe，其中包含相同的行数。之后，我将该数据帧改回RDD，并应用不同的映射逻辑来转换行的字段(映射时不应用过滤器)。然而，输出数据帧不包含相同数量的行-它们在没有任何解释的情况下被丢弃。(OUTPUT_FORMAT).optio
浏览 6提问于2019-07-19得票数 0
1回答
火花删除行
、、、、
我有一个包含大约20k行的DataFrame。为了理解上下文--我正在测试缺失数据的分类模型，每一行都有一个unix时间戳。186行对应于3秒(每秒有62行数据)。
我的目标是，当数据正在流时，数据很可能会丢失几秒钟。我正在从一个时间窗口中提取特性，所以我想看看缺少的数据对模型性能的影响。我认
浏览 4提问于2016-08-08得票数 1
回答已采纳
1回答
videos.csv如下所示099acca-8888-48ca,Action,Comedy火花Scala
val
浏览 2提问于2016-12-29得票数 3
1回答
我对Spark完全陌生，目前我正在尝试使用Python编写一段简单的代码，对一组数据执行KMeans操作。/spark/python/lib/pyspark.zip/pyspark/serializers.py"，第263行中，在dump_stream vs =list(迭代器，itertools.islice更新：@Garren我得到的错误是：
我得到的错误是：>>> kmm =
浏览 2提问于2017-03-02得票数 5
回答已采纳
1回答
我正在使用做一些工作，并有一个包含在每一行中的重复事务示例的rdd。这将导致模型训练函数因这些重复项而引发错误。我对Spark相当陌生，我想知道如何删除rdd行中的重复项。例如： from pyspark.mllib.fpm import FPGrowth
data = [["a", "a", "b", &
浏览 2提问于2016-09-06得票数 0
回答已采纳
1回答
我试图用CDH在纱线上运行电火花，在这个生火的主程序中，有一条类似于rdd.pipe("XXX.sh")的语句，每次我运行它时，都会弹出一个被拒绝的错误，我该怎么办来解决这个错误呢？谢谢。/pyspark/rdd.py"，第2346行，pipeline_func文件"/data/yarn/nm/usercache/work/appcache/
浏览 3提问于2017-05-25得票数 0
1回答
在我当前的pyspark项目中，我正在将一个S3文件读入RDD，并对其运行一些简单的转换。下面是代码。一旦我得到它，我运行最后一个过滤器，如图所示，只拾取字段5中value = 1的行，到目前为止一切都很好。 接下来，我想使用模式将segmentsRDD转换为DF，如下所示。start_offset","end_offset","time_shifted","day_shifted"
浏览 20提问于2020-01-30得票数 0
回答已采纳
但是，根据文档，即使这个函数显然存在于吡火花RDD类中，我也无法使用它并得到以下错误：我在下面一行中调用后一个函数：进口商品如下：from pysp
浏览 2提问于2018-10-28得票数 4
回答已采纳
1回答
我有一个pyspark RDD，它有大约200万个元素。我不能一次收集它们，因为它会导致OutOfMemoryError异常。 如何批量采集？这是一个潜在的解决方案，但我怀疑还有更好的方法:收集一个批处理(使用take、https://spark.apache.org/docs/3.1.2/api/python/reference/api/pyspark.RDD.take.html#pysp
浏览 29提问于2021-10-12得票数 0
回答已采纳
1回答
我正在尝试使用mongoDB连接器在SPARK中执行python文件。python文件执行一个查询，从mongoDB获取一些数据，然后他们使用星火中的映射操作来处理这些数据。，在bad_reviews = reviews_1.rdd.map(lambda r: r.text).collect()文件"/usr/local/spark/python/lib/pyspark.zip/pyspark/rdd</em
浏览 4提问于2016-11-27得票数 4
我正在使用Jupyter notebook，刚刚开始学习Apache spark，但在创建简单的RDD时遇到错误：错误是：，每种情况下都会发生这种情况
浏览 3提问于2017-03-30得票数 5
我正在使用ApacheSpark1.6.2但是，我必须首先将它转换为RDD来进行映射，以获得我想要的数据(列)。_inferSchema File "c:\spark\python\lib\pysp
浏览 3提问于2017-01-14得票数 4
回答已采纳
6回答
如何在PySpark中从RDD中删除行？特别是第一行，因为这往往在我的数据集中包含列名。通过仔细阅读API，我似乎找不到一种简单的方法来做到这一点。当然，我可以通过Bash / HDFS来完成这个任务，但我只想知道这是否可以在PySpark内部完成。
浏览 5提问于2014-07-13得票数 28
回答已采纳
在scala中，当我有一个类似于：List(("a",1),("a",2),("b",3),("b",4),("b",5),("a",6))的RDD列表时，我想要计算每个字符的avg数。/pyspark/python/lib/pyspark.zip/pyspark/worker.py"，第611<em
浏览 5提问于2021-12-24得票数 0
回答已采纳
我正在Spark中创建Row对象。我不希望我的字段按字母顺序排序。但是，如果我这样做，它们是按字母顺序排序的。row = Row(foo=1, bar=2)Row(bar=2, foo=1)
然后，当我在这个对象上创建一个dataframe时，列的顺序将是bar优先，foo第二，而我更喜欢的是相反的顺序。我知道我可以使用"_1“和"_2”(分别表示"foo“和"bar”)，然后分配一个模式(使用
浏览 1提问于2016-02-11得票数 14
回答已采纳
我正在尝试运行这个函数def iterateRDD(record, model):(lambda行:iterateRDD(行，模型))文件"C:\Python\Python27\lib\site-packages\pyspark\rdd.py"，第789行中，在foreach self.mapPartitions(processPa
浏览 2提问于2019-07-17得票数 0
我想回答的问题是：使用映射创建列指定的NumPy数组的RDD。RDD的名称将是行。我的代码：Rows = df.select(col).rdd.map(make_array)
输入这个之后，我会得到一个奇怪的错误，它基本上是这样的: PySpark在worker中有不同的版本2.7与在驱动程序3.6中有不同的版本，
浏览 2提问于2019-07-04得票数 2
回答已采纳
我有一个pyspark RDD (myRDD)，它是一个可变长度的ID列表，比如我有一个包含列ID和value的pyspark数据框架(myDF)。myDF.select(F.collect_set("value&qu
浏览 26提问于2017-08-24得票数 0
我知道Kmeans不是一个适用于分类数据的很好的选择，但是我们在spark 1.4中没有太多的选择来对分类数据进行聚类。不管上面的问题。我在下面的代码中有错误。所有的评论都被删除了，谢谢你的帮助！我得到的错误："/usr/hdp/2.3.2.0-2950/spark/python/lib/pyspark.zip/pyspark/serializers.p
浏览 1提问于2016-02-02得票数 0
我正在学习“学习PySpark”教程(在)。：
< class 'pyspark.rdd.PipelinedRDD‘>
< class‘pyspk.rdd.rdd’>"，第139行，迭代器中用于obj的dump_stream :文件"/content/spark-2.3.1-bin-hadoop2.7&#x
浏览 0提问于2018-08-08得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
