如何访问 DataFrame 中由 Row 创建 DataFrame 的列答案

【问题标题】：How to access column in Dataframe where DataFrame is created by Row如何访问 DataFrame 中由 Row 创建 DataFrame 的列
【发布时间】：2020-04-13 23:24:02
【问题描述】：

我是 pyspark 的新手
想从由 Row 创建的 DataFrame 访问列。
请参阅我的 .py 文件中的以下代码
它抛出错误 AttributeError: 'DataFrame' object has no attribute 'product'

import findspark

findspark.init("/opt/spark")

from pyspark.sql import SparkSession
from pyspark.sql import Row
from pyspark.sql import SQLContext



productRevenue = Row("product", "category", "revenue")
spark = SparkSession \
    .builder \
    .appName("DataFrame Learning") \
    .getOrCreate()

sqlContext = SQLContext(spark)

productRevenue1 = productRevenue("product", "Cell phone", 6000)
productRevenue2 = productRevenue("Normal", "Tablet", 1500)
productRevenue3 = productRevenue("Mini", "Tablet", 5500)
productRevenue4 = productRevenue("Ultra thin", "Cell phone", 5000)
productRevenue5 = productRevenue("Very thin", "Cell phone", 6000)
productRevenue6 = productRevenue("Big", "Tablet", 2500)
productRevenue7 = productRevenue("Bendable", "Cell phone", 3000)
productRevenue8 = productRevenue("Foldable", "Cell phone", 3000)
productRevenue9 = productRevenue("Pro", "Tablet", 5500)
productRevenue10 = productRevenue("Pro2", "Tablet", 5500)

productRevenueAll = Row(
    productRevenue=[productRevenue1, productRevenue2, productRevenue3, productRevenue4, productRevenue5,
                    productRevenue6, productRevenue7, productRevenue8, productRevenue9, productRevenue10])

dataFrame = spark.createDataFrame(productRevenueAll)



filter_df = dataFrame.filter((dataFrame.product=="product") )

【问题讨论】：

标签： apache-spark pyspark

【解决方案1】：

要从 Rows 创建 DataFrame，一种方法是在 Rows 列表上调用 SparkSession.createDataFrame()。

如果你想创建类似的DataFrame

# +----------+----------+-------+
# |   product|  category|revenue|
# +----------+----------+-------+
# |   product|Cell phone|   6000|
# |    Normal|    Tablet|   1500|
# |      Mini|    Tablet|   5500|
# |             ...             |
# +----------+----------+-------+

# with Schema:

# root
#  |-- product: string (nullable = true)
#  |-- category: string (nullable = true)
#  |-- revenue: long (nullable = true)

然后，不要将productRevenueAll 作为行的行，而是将其更改为行列表，例如：

productRevenueAll = [
    productRevenue1, productRevenue2, productRevenue3, 
    productRevenue4, productRevenue5, productRevenue6, 
    productRevenue7, productRevenue8, productRevenue9,
    productRevenue10,
]

dataFrame = spark.createDataFrame(productRevenueAll)

# then use it like:

dataFrame.product
# Column<b'product'>

dataFrame.select(dataFrame.product).show()
# +----------+
# |   product|
# +----------+
# |   product|
# |    Normal|
# |      Mini|
# |   ...    |
# +----------+

但是，如果您真的打算创建一个嵌套结构，例如：

# +-----------------------------+
# |      productRevenue         |
# +----------+----------+-------+
# |   product|  category|revenue|
# +----------+----------+-------+
# |   product|Cell phone|   6000|
# |    Normal|    Tablet|   1500|
# |             ...             |
# +----------+----------+-------+

# with Schema:

# root
#  |-- productRevenue: array (nullable = true)
#  |    |-- element: struct (containsNull = true)
#  |    |    |-- product: string (nullable = true)
#  |    |    |-- category: string (nullable = true)
#  |    |    |-- revenue: long (nullable = true)

为createDataFrame() 提供一项列表，例如：

productRevenueAllNested = Row(
    productRevenue=[
        productRevenue1, productRevenue2, productRevenue3, 
        productRevenue4, productRevenue5, productRevenue6, 
        productRevenue7, productRevenue8, productRevenue9, 
        productRevenue10,
    ])

dataFrameNested = spark.createDataFrame([productRevenueAllNested]) 

# then access it like
dataFrameNested.printSchema()

dataFrameNested.select(dataFrameNested.productRevenue).show()
# +----------------------+
# |productRevenue.product|
# +----------------------+
# |  [product, Normal,...|
# +----------------------+

【讨论】：

感谢@Quar！我尝试将 productRevenueAll 作为 Row of Rows，将其更改为为我工作的 Rows 列表！不知道为什么问题降级了 wtth -1 的声誉:-)
@GaurangPopat 很高兴它有帮助：D

【解决方案2】：

您正在嵌套 Row 对象，这会导致产生 struct 字段。

您可以通过以下方式从Row 对象createDataFrame。

Example:

#using .toDF to create dataframe
sc.parallelize([productRevenue1, productRevenue2, productRevenue3, productRevenue4, productRevenue5,productRevenue6, productRevenue7, productRevenue8, productRevenue9, productRevenue10]).toDF().show()

#using spark.createDataFrame to create dataframe
spark.createDataFrame([productRevenue1, productRevenue2, productRevenue3, productRevenue4, productRevenue5,productRevenue6, productRevenue7, productRevenue8, productRevenue9, productRevenue10]).show()

#creating dataframe from rdd
productRevenue=sc.parallelize([productRevenue1, productRevenue2, productRevenue3, productRevenue4, productRevenue5,productRevenue6, productRevenue7, productRevenue8, productRevenue9, productRevenue10])

#creating dataframe from list
productRevenue=[productRevenue1, productRevenue2, productRevenue3, productRevenue4, productRevenue5,productRevenue6, productRevenue7, productRevenue8, productRevenue9, productRevenue10]

spark.createDataFrame(productRevenue).show()
#+----------+----------+-------+
#|   product|  category|revenue|
#+----------+----------+-------+
#|   product|Cell phone|   6000|
#|    Normal|    Tablet|   1500|
#|      Mini|    Tablet|   5500|
#|Ultra thin|Cell phone|   5000|
#| Very thin|Cell phone|   6000|
#|       Big|    Tablet|   2500|
#|  Bendable|Cell phone|   3000|
#|  Foldable|Cell phone|   3000|
#|       Pro|    Tablet|   5500|
#|      Pro2|    Tablet|   5500|
#+----------+----------+-------+


dataFrame=spark.createDataFrame(productRevenue)

dataFrame.filter((dataFrame.product=="product") ).show()
#+-------+----------+-------+
#|product|  category|revenue|
#+-------+----------+-------+
#|product|Cell phone|   6000|
#+-------+----------+-------+

【讨论】：