在PySpark中创建类的Apache Spark RDD

Question

我必须将Scala代码转换为python。

scala代码将string的RDD转换为case类的RDD。代码如下：

case class Stock(
                  stockName: String,
                  dt: String,
                  openPrice: Double,
                  highPrice: Double,
                  lowPrice: Double,
                  closePrice: Double,
                  adjClosePrice: Double,
                  volume: Double
                )


  def parseStock(inputRecord: String, stockName: String): Stock = {
    val column = inputRecord.split(",")
    Stock(
      stockName,
      column(0),
      column(1).toDouble,
      column(2).toDouble,
      column(3).toDouble,
      column(4).toDouble,
      column(5).toDouble,
      column(6).toDouble)
  }

  def parseRDD(rdd: RDD[String], stockName: String): RDD[Stock] = {
    val header = rdd.first
    rdd.filter((data) => {
      data(0) != header(0) && !data.contains("null")
    })
      .map(data => parseStock(data, stockName))
  }  

可以在PySpark中实现吗？我尝试使用以下代码，但出现错误

from dataclasses import dataclass

@dataclass(eq=True,frozen=True)
class Stock:
    stockName : str
    dt: str
    openPrice: float
    highPrice: float
    lowPrice: float
    closePrice: float
    adjClosePrice: float
    volume: float


 

def parseStock(inputRecord, stockName):
  column = inputRecord.split(",")
  return Stock(stockName,
               column[0],
               column[1],
               column[2],
               column[3],
               column[4],
               column[5],
               column[6])

def parseRDD(rdd, stockName):
  header = rdd.first()
  res = rdd.filter(lambda data : data != header).map(lambda data : parseStock(data, stockName))
  return res

错误Py4JJavaError：调用z:org时出错。阿帕奇。火花应用程序编程接口。蟒蛇蟒蛇。收集和服务：组织。阿帕奇。火花SparkException：作业因阶段失败而中止：阶段21.0中的任务0失败1次，最近的失败：阶段21.0中的任务0.0丢失（TID 31，localhost，executor driver）：org。阿帕奇。火花应用程序编程接口。蟒蛇PythonException：回溯（最近一次调用上次）：

文件“/content/spark-2.4.5-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/worker.py”，第364行，在main func、profiler、反序列化器、serializer=read\u命令（pickleSer、infle）文件“/content/spark-2.4.5-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/worker.py”，第69行，在read\u命令command=serializer中_read\u with\u length（file）file“/content/spark-2.4.5-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/serializers.py”，第173行，在\u read\u with\u length return self中。loads（obj）File“/content/spark-2.4.5-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/serializers.py”，第587行，在loads return pickle中。加载（obj，encoding=encoding）AttributeError：无法在上获取属性“main”

Answer 1

数据集API不适用于python。

“数据集是一个分布式数据集合。数据集是Spark 1.6中添加的一个新接口，它提供了RDD（强类型、使用强大lambda函数的能力）的优点，以及Spark SQL优化的执行引擎的优点。数据集可以从JVM对象构建，然后使用函数转换（map、flatMap、filter等）进行操作. 数据集API有Scala和Java两种版本。Python不支持Dataset API。但由于Python的动态特性，Dataset API的许多优点已经可用（即，您可以按名称自然访问行的字段row.columnName）。R的情况类似。"

https://spark.apache.org/docs/latest/sql-programming-guide.html