pyspark

0热度

1回答

from pyspark.sql import Row, functions as F row = Row("UK_1","UK_2","Date","Cat") agg = '' agg = 'Cat' tdf = (sc.parallelize ([ row(1,1,'12/10/2016',"A"), row(1,2,None,'A'),

1热度

1回答

访问循环中的RDD时发生火花 - 酸洗错误

我在Spark中实现了k-means算法。当我运行下面的代码时，出现酸洗错误（如下所示）。如果我修改它并将所有内容放在循环之外，它将正确计算质心。 sc = SparkContext(appName="Document Similarity") lines = sc.wholeTextFiles(sys.argv[1]) articles = lines.flatMap(lambda x:

1热度

1回答

如何获取pyspark数据框中具有最大值的列的名称

我们如何获得列pyspark数据框的名称？ Alice Eleonora Mike Helen MAX 0 2 7 8 6 Mike 1 11 5 9 4 Alice 2 6 15 12 3 Eleonora 3 5 3 7 8 Helen 我需要这样的东西。没有的最高值，我能够得到的最高值列的名字，我需要的名字

0热度

1回答

pyspark数据帧与上一行的复杂计算

我正在使用Pyspark并试图弄清楚如何使用以前的列进行复杂计算。我认为通常有两种方法可以使用以前的列进行计算：Windows和mapwithPartition。我认为我的问题太复杂，无法通过windows解决，我想将结果视为一个判断行，而不是列。所以我试图使用mapwithpartition。我在这种语法方面遇到了麻烦。例如，这是一个粗略的代码草稿。 def change_dd(rows):

2热度

1回答

如何在Hadoop HDFS上存储数千个CSV文件

情况如何？我有一万个CSV文件（每个250kb - 270kb），我想用Spark（Pyspark精确地）处理。每个CSV文件都代表一个特定事件的过程数据。你可以说一个CSV文件代表一个对象。由于我想将数据存储在HDFS上，所以我必须找到连接数据的方式（因为在HDFS上存储大量微小数据效率不高）。一个CSV文件的片段（简化）。 Time Module v1 v2 v3 v4 v5 v6

-1热度

1回答

pyspark：DataFrame.withColumn（）有时需要以不同的名称分配给新的DataFrame

这似乎特别发生在我将数字列乘以标量时，将DataFrame写回HDFS，然后尝试查看当我再次将它加载到DataFrame中时的值。例如，在pyspark shell中不会发生。 df = df.withColumn('AMOUNT', df.AMOUNT*lit(-1)) =>不翻转列 df_new = df.withColumn('AMOUNT', df.AMOUNT*lit(-1)) =>作品

1热度

1回答

是否有可能将Scala方法转换为python方法

我想知道我们是否可以在运行时将Scala中定义的方法转换为Python中的方法。我正在学习Apache Spark Scala API作为我的项目的一部分，并寻找将Scala方法转换为Python语言可识别格式的方法。我的目标是将方法从Scala程序发送到外部Python程序，Python程序应该能够使用或执行它。 Scala代码=> LAMBDA line => line.split(" ")

0热度

1回答

如何使用pyspark运行并行程序？

我想用我们的Spark集群并行运行程序。我的想法是做某事如下所示： def simulate(): #some magic happening in here return 0 spark = ( SparkSession.builder .appName('my_simulation') .enableHiveSupport() .getOrCrea

0热度

1回答

变换RDD到有效的输入，使用包含与下面的代码CSV文件的目录的火花mllib算法k均值

我计算TF和IDF： import argparse from os import system ### args parsing parser = argparse.ArgumentParser(description='runs TF/IDF on a directory of text docs') parser.add_argument("-i","--input", help

0热度

1回答

Pyspark Split列

from pyspark.sql import Row, functions as F row = Row("UK_1","UK_2","Date","Cat",'Combined') agg = '' agg = 'Cat' tdf = (sc.parallelize ([ row(1,1,'12/10/2016',"A",'Water^World'),