首页 > TAG信息列表 > spark-dataframe

PySpark 1.6：DataFrame：将一列从字符串转换为浮点/双精度

在PySpark 1.6 DataFrame中,当前没有Spark内置函数可以将字符串转换为float / double. 假设我们有一个带有(‘house_name’,’price’)的RDD,两个值都作为字符串.您想将价格从字符串转换为浮动. 在PySpark中,我们可以应用map和python float函数来实现这一点. New_RDD = RawDataRD

Python Spark DataFrame：用SparseVector替换null

在Spark中,我有以下名为“ df”的数据框,其中包含一些空条目： +-------+--------------------+--------------------+ | id| features1| features2| +-------+--------------------+--------------------+ | 185|(5,[0,1,4],[0.1,

python-如何使用“] | [”分隔符读取pyspark中的文件

数据如下所示： pageId] | [page] | [Position] | [sysId] | [carId  0005] | [宝马] | [南部] | [AD6] | [OP4 至少有50列和数百万行. 我确实尝试使用下面的代码来阅读： dff = sqlContext.read.format(“ com.databricks.spark.csv”).option(“ header”,“ true”).option(“ infe

python-PySpark-从Numpy矩阵创建DataFrame

我有一个numpy的矩阵： arr = np.array([[2,3], [2,8], [2,3],[4,5]]) 我需要从arr创建一个PySpark数据框.我无法手动输入值,因为arr的长度/值将动态变化,因此我需要将arr转换为数据帧. 我尝试以下代码未成功. df= sqlContext.createDataFrame(arr,["A", "B"]) 但是,出现以下错误

pyspark dataframe,groupby和计算列的方差

我想对pyspark数据框进行分组并计算特定列的方差.一般而言,这很容易,可以像这样完成 from pyspark.sql import functions as func AVERAGES=df.groupby('country').agg(func.avg('clicks').alias('avg_clicks')).collect() 但是对于方差,在功能子模块中似乎没有任何聚合函数(我也

python-如何在不使用RDD API的情况下摆脱pyspark数据帧中的行包装器对象？

我针对临时视图发布以下SQL语句 cloudantdata.createOrReplaceTempView("washingflat") sqlDF = spark.sql("SELECT temperature FROM washingflat") sqlDF.rdd.map(lambda row : row.temperature).collect() 我只是对普通的(展开的)整数值感兴趣.到目前为止,我使用dataframe AP

如何将json文件读入python？

我是JSON和Python的新手,对此的任何帮助都将非常感激. 我读到了json.loads,但很困惑 如何使用json.loads将文件读入Python？ 以下是我的JSON文件格式： { "header": { "platform":"atm" "version":"2.0" } "details":[

python – 将Pandas数据帧转换为Spark数据帧错误

我正在尝试将Pandas DF转换为Spark. DF头： 10000001,1,0,1,12:35,OK,10002,1,0,9,f,NA,24,24,0,3,9,0,0,1,1,0,0,4,543 10000001,2,0,1,12:36,OK,10002,1,0,9,f,NA,24,24,0,3,9,2,1,1,3,1,3,2,611 10000002,1,0,4,12:19,PA,10003,1,1,7,f,NA,74,74,0,2,15,2,0,2,3,1,2,2,691 码： d

python – 如何在pySpark数据帧中添加行ID [复制]

参见英文答案 > Primary keys with Apache Spark                                    3个 我有一个csv文件;我在pyspark中转换为DataFrame(df);经过一番改造;我想在df中添加一列;这应该是简单的行id(从0或1开始到N). 我在rdd中转换了df并使用

Spark RDD到DataFrame python

我试图将Spark RDD转换为DataFrame.我已经看到了传递方案的文档和示例 sqlContext.CreateDataFrame(rdd,schema)函数. 但我有38列或字段,这将进一步增加.如果我手动给出指定每个字段信息的模式,那将会是如此繁琐的工作. 有没有其他方法可以在不知道先前列的信息的情况下指定模式.解

java – 使用spark streaming从数据库中读取流

我想使用spark streaming来读取来自RDBMS数据库的数据,比如mysql. 但我不知道如何使用JavaStreamingContext来做到这一点 JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.milliseconds(500)); DataFrame df = jssc. ?? 我在互联网上搜索,但我没有找到任

如何使用Java将unix epoch的列转换为Apache spark DataFrame中的Date？

我有一个json数据文件,其中包含一个属性[creationDate],它是“long”数字类型的unix epoc. Apache Spark DataFrame架构如下所示： root |-- creationDate: long (nullable = true) |-- id: long (nullable = true) |-- postTypeId: long (nullable = true) |-- tags: array

python – Pyspark：TaskMemoryManager：无法分配页面：在错误分析中需要帮助

我在独立群集模式下运行spark作业时遇到这些错误. 我的火花工作旨在： >运行一些groupby, >数,>并加入以获得最终的df,然后加入df.toPandas().to_csv(). 输入数据集为524 Mb.我得到的错误： WARN TaskMemoryManager: Failed to allocate a page (33554432 bytes), try again. 多次重

python – PySpark：添加一个新列,其中包含从列创建的元组

这里我创建了一个dateframe,如下所示, df = spark.createDataFrame([('a',5,'R','X'),('b',7,'G','S'),('c',8,'G','S')], ["Id","V1","V

python – PySpark Dataframe：逗号点

我在浮点数中使用逗号导入数据,我想知道如何将’逗号转换为点.我正在使用pyspark数据帧,所以我尝试了这个： commaToDot = udf(lambda x : str(x).replace(',', '.'), FloatType()) myData.withColumn('area',commaToDot(myData.area)) 它绝对不起作用. 那么我们可以直接用spark替

加入一个数据帧spark java

首先,感谢您抽出时间阅读我的问题. 我的问题如下：在Spark with Java中,我在两个数据帧中加载了两个csv文件的数据. 这些数据框将具有以下信息. Dataframe机场 Id | Name | City ----------------------- 1 | Barajas | Madrid Dataframe airport_city_state City | state ---

python – 用户定义的函数打破了pyspark数据帧

我的火花版是1.3,我正在使用pyspark. 我有一个名为df的大型数据框. from pyspark import SQLContext sqlContext = SQLContext(sc) df = sqlContext.parquetFile("events.parquet") 然后,我选择数据帧的几列,并尝试计算行数.这很好用. df3 = df.select("start", "end", "mrt") p

如何使用apache spark java中的hadoop office库将数据集写入excel文件

目前我正在使用com.crealytics.spark.excel来读取excel文件,但是使用这个库我无法将数据集写入excel文件. 这个link说使用hadoop办公室库(org.zuinnote.spark.office.excel)我们可以读写excel文件 请帮我把数据集对象写入spark java中的excel文件.解决方法:您可以使用org.zuinnote.

使用Java在Apache Spark中复制数据集中的一行n次

我试图从数据集中复制一行n次并从中创建一个新的数据集.但是,在复制时我需要为每个复制更改列的值,因为它最终会在最终存储时作为主键. 以下是来自SO post：Replicate Spark Row N-times的Scala代码 import org.apache.spark.sql.functions._ val result = singleRowDF .withColu

Spark Dataframes：如何更改Java / Scala中的列顺序？

在加入两个数据帧之后,我发现列顺序已经改变了我认为的那样. 例如：在b上与列[b,c,d,e]和[a,b]连接两个数据帧产生[b,a,c,d,e]的列顺序. 如何更改列的顺序(例如,[a,b,c,d,e])？ 我已经找到了在Python / R中实现它的方法,但不是Scala或Java.是否有任何方法允许交换或重新排序dataframe列

python – 将GraphFrames ShortestPath Map转换为PySpark中的DataFrame行

我试图找到最有效的方法从GraphFrames函数shortestPaths获取Map输出,并将每个顶点的距离映射平铺为新DataFrame中的各个行.通过将距离列拉入字典然后从那里转换为pandas数据帧然后转换回Spark数据帧,我已经能够非常笨拙地做到这一点,但我知道必须有更好的方法. from graphframes i

mysql – 在“GROUP BY”子句中重用select表达式的结果？

在MySQL中,我可以有这样的查询： select cast(from_unixtime(t.time, '%Y-%m-%d %H:00') as datetime) as timeHour , ... from some_table t group by timeHour, ... order by timeHour, ... 其中GROUP BY中的timeHour是select表达式的结果. 但我只是尝试

python – PySpark中具有多个列的日期算法

我正在尝试使用PySpark数据框中的多个列进行一些中等复杂的日期算术.基本上,我有一个名为number的列,表示我需要过滤的created_at时间戳之后的周数.在PostgreSQL中你可以乘以interval based on the value in a column,但我似乎无法使用SQL API或Python API弄清楚如何在PySpark中执