标签: pyspark-sql

我来自pandas背景,习惯于将CSV文件中的数据读入数据帧,然后使用简单命令将列名更改为有用的东西:

df.columns = new_column_name_list

但是,在使用sqlContext创建的pyspark数据帧中,这同样不起作用.我可以轻松解决的唯一解决方案如下:

df = sqlContext.read.format("com.databricks.spark.csv").options(header='false', inferschema='true', delimiter='\t').load("data.txt")
oldSchema = df.schema
for i,k in enumerate(oldSchema.fields):
  k.name = new_column_name_list[i]
df = sqlContext.read.format("com.databricks.spark.csv").options(header='false', delimiter='\t').load("data.txt", schema=oldSchema)

这基本上是定义变量两次并首先推断模式然后重命名列名,然后再次使用更新的模式加载数据帧.

像熊猫一样,有更好更有效的方法吗？

我的火花版是1.5.0

请为Pandas建议pyspark数据帧替代方案df['col'].unique().

我想列出pyspark数据帧列中的所有唯一值.

不是SQL类型的方式(registertemplate然后SQL查询不同的值).

我也不需要groupby->countDistinct,而是想检查该列中的不同VALUES.

我有一个日期pyspark数据帧,其格式为字符串列MM-dd-yyyy,我试图将其转换为日期列.

我试过了:

df.select(to_date(df.STRING_COLUMN).alias('new_date')).show()

我得到一串空值.有人可以帮忙吗？

我花了相当多的时间阅读pyspark和spark- dataframe标签的一些问题,而且我经常发现海报没有提供足够的信息来真正理解他们的问题.我经常评论要求他们发布MCVE,但有时让他们显示一些样本输入/输出数据就像拔牙一样.例如:请参阅有关此问题的评论.

也许问题的一部分是人们只是不知道如何轻松地为火花数据帧创建MCVE.我认为将这个pandas问题的spark-dataframe版本作为可以链接的指南是有用的.

那么如何创造一个好的,可重复的例子呢？

假设我有一个火花数据帧df1,有几列(其中列'id')和数据帧df2有两列,'id'和'other'.

有没有办法复制以下命令

sqlContext.sql("SELECT df1.*, df2.other FROM df1 JOIN df2 ON df1.id = df2.id")

通过仅使用诸如join(),select()之类的pyspark函数？

我必须在函数中实现此连接,并且我不希望强制将sqlContext作为函数参数.

谢谢!

我见过各种各样的人都认为这Dataframe.explode是一种有用的方法,但它会导致比原始数据帧更多的行,这根本不是我想要的.我只是想做Dataframe相当于非常简单:

rdd.map(lambda row: row + [row.my_str_col.split('-')])

它看起来像:

col1 | my_str_col
-----+-----------
  18 |  856-yygrm
 201 |  777-psgdg

并将其转换为:

col1 | my_str_col | _col3 | _col4
-----+------------+-------+------
  18 |  856-yygrm |   856 | yygrm
 201 |  777-psgdg |   777 | psgdg

我知道pyspark.sql.functions.split(),但它导致嵌套数组列而不是我想要的两个顶级列.

理想情况下,我希望这些新列也可以命名.

在pyspark 1.6.2中,我可以导入col函数

from pyspark.sql.functions import col

但是当我尝试在Github源代码中查找它时,我发现文件中没有col函数,functions.pypython如何导入一个不存在的函数？

在pandas中,这可以通过column.name来完成.

但是当它的火花数据帧列如何做同样的事情？

例如,调用程序有一个spark数据帧:spark_df

>>> spark_df.columns
['admit', 'gre', 'gpa', 'rank']

这个程序调用我的函数:my_function(spark_df ['rank'])在my_function中,我需要列的名称即'rank'

如果是pandas数据帧,我们可以在my_function中使用

>>> pandas_df['rank'].name
'rank'

我想在Spark数据帧上计算组分位数(使用PySpark).近似或精确的结果都可以.我更喜欢在groupBy/ 的上下文中使用的解决方案agg,以便我可以将它与其他PySpark聚合函数混合使用.如果由于某种原因这是不可能的,那么不同的方法也可以.

这个问题是相关的,但没有说明如何approxQuantile用作聚合函数.

我也可以访问percentile_approxHive UDF,但我不知道如何将它用作聚合函数.

为了特异性,假设我有以下数据帧:

from pyspark import SparkContext
import pyspark.sql.functions as f

sc = SparkContext()    

df = sc.parallelize([
    ['A', 1],
    ['A', 2],
    ['A', 3],
    ['B', 4],
    ['B', 5],
    ['B', 6],
]).toDF(('grp', 'val'))

df_grp = df.groupBy('grp').agg(f.magic_percentile('val', 0.5).alias('med_val'))
df_grp.show()

预期结果是:

+----+-------+
| grp|med_val|
+----+-------+
|   A|      2|
|   B|      5|
+----+-------+

import numpy as np

df = spark.createDataFrame(
    [(1, 1, None), (1, 2, float(5)), (1, 3, np.nan), (1, 4, None), (1, 5, float(10)), (1, 6, float('nan')), (1, 6, float('nan'))],
    ('session', "timestamp1", "id2"))

预期产出

每列的数量为nan/null的数据帧

注意: 我在堆栈溢出中发现的先前问题仅检查null而不是nan.这就是为什么我创造了一个新问题.

我知道我可以在spark中使用isnull()函数来查找Spark列中的Null值的数量但是如何在Spark数据帧中找到Nan值？