在 Spark 中，我如何通过名称本身而不是索引来读取字段

Question

我使用 Spark 1.3。

我的数据有 50 多个属性，因此我选择了自定义 class。

如何通过名称而不是位置

从自定义 Class 访问字段

这里每次我需要调用一个方法productElement(0)

此外，我不应该使用 case class，因此我使用自定义 class 作为架构。

 class OnlineEvents(gsm_id:String,
          attribution_id:String,
          event_date:String,
          event_timestamp:String,
          event_type:String
          ) extends Product {

  override def productElement(n: Int): Any = n match {
  case 0 => impression_id
  case 1 => attribution_id
  case 2 => event_date
  case 3 => event_timestamp
  case 4 => event_type

  case _ => throw new IndexOutOfBoundsException(n.toString)
 }

  override def productArity: Int = 5

  override def canEqual(that: Any): Boolean = that.isInstanceOf[OnlineEvents]

 }

我的 Spark 代码：

  val onlineRDD = sc.textFile("/user/cloudera/input_files/online_events.txt")

  val schemaRDD = onlineRDD.map(record => {
                                         val arr: Array[String] = record.split(",")
                                          new OnlineEvents(arr(0),arr(1),arr(2),arr(3),arr(4))
})
 val keyvalueRDD =  schemaRDD .map(online => ((online.productElement(0).toString,online.productElement(4).toString),online))

如果我尝试从 OnlineEvents 访问任何字段，那么我需要使用 productElement() 。（即 online.productElement(0) 用于 gsm_id ）

我可以直接访问该字段作为 online.gsm_id ... online.event_type ，以便我的代码易于阅读

当我对模式使用自定义 Class 时，如何通过名称直接访问字段？

Answer 1

根据我对你问题的理解，你需要在OnlineEvents到return里面定义一些functions的类型。所以你的解决方案应该是

class OnlineEvents(gsm_id:String,
                   attribution_id:String,
                   event_date:String,
                   event_timestamp:String,
                   event_type:String
                  ) extends Product {
  def get_gsm_id(): String ={
    gsm_id
  }

  def get_attribution_id(): String ={
    attribution_id
  }

  def get_event_date(): String ={
    event_date
  }

  def get_event_timestamp(): String ={
    event_timestamp
  }

  def get_event_type(): String ={
    event_type
  }

  override def productElement(n: Int): Any = n match {
    case 0 => gsm_id
    case 1 => attribution_id
    case 2 => event_date
    case 3 => event_timestamp
    case 4 => event_type

    case _ => throw new IndexOutOfBoundsException(n.toString)
  }

  override def productArity: Int = 5

  override def canEqual(that: Any): Boolean = that.isInstanceOf[OnlineEvents]

}

并调用如下函数

val keyvalueRDD =  schemaRDD .map(online => ((online.get_gsm_id().toString,online.get_event_type().toString),online))

Answer 2

我强烈建议每个用例使用一个案例 class（它们一起涵盖了使用数据的所有用例）。

单个用例将是单个用例class，这将使您无需考虑如何维护 50 多个字段。

是的，对于 10 个 5 字段案例 class，您 "trade" 一个 50 或更多字段的大字段 class，但考虑到它是多么容易创建一个案例 class 以及他们如何很好地描述您的数据我认为值得麻烦。

在 Spark 中，我如何通过名称本身而不是索引来读取字段

In Spark How do i read a field by its name itself instead by its index

apache-spark

apache-spark-1.3