根據基于嵌套ID的串列乘以事件案例類-有解無憂

賞金將在 5 小時后到期。此問題的答案有資格獲得 50聲望賞金。 CoolGoose正在從有信譽的來源尋找答案：

基本上，我一次處理一個事件并回傳一個案例類。根據嵌套的 ID 值，一條記錄如何乘以多行。所以不知道該怎么做。

我正在處理一個資料幀并使用事件案例類轉換為 Dataset[Event]。但是有嵌套的 Id，我需要根據嵌套設備的扁平化來乘以事件：os。

我能夠在 Kafka 事件級別回傳案例類 Event。但不確定如何增加事件。

卡夫卡傳入事件：

{
  "partition": 1,
  "key": "34768_20220203_MFETP501",
  "offset": 1841543,
  "createTime": 1646041475348,
  "topic": "topic_int",
  "publishTime": 1646041475344,
  "errorCode": 0,
  "userActions": {
    "productId": "3MFETP501",
    "createdDate": "2022-02-26T11:19:35.786Z",
    "events": [
      {
        "GUID": "dbb1-f38b-f7f0-44af-90da-80179412f89c",
        "eventDate": "2022-02-26T11:19:35.786Z",
        "familyId": 2010,
        "productTypeId": 1004678,
        "serialID": "890479804",
        "productName": "MFE Total Protection 2021 Family Pack",
        "features": {
          "mapping": [
            {
              "deviceId": 999795,
              "osId": [
                100
              ]
            },
            {
              "deviceId": 987875
              "osId": [
                101
              ]
            }
          ]
        }
      }
    ]
  }
}

Event 的預期輸出案例類

Event("3MFETP501","1004678","2010","3MFETP501:890479804","MFE Total Protection 2021 Family Pack","999795_100", Map("targetId"->"999795_100") )
Event("3MFETP501","1004678","2010","3MFETP501:890479804","MFE Total Protection 2021 Family Pack","987875_100", Map("targetId"->"987875_100") )

case class Event(
                    productId: String,
                    familyId: String,
                    productTypeId: String,
                    key: String,
                    productName: String,
                    deviceOS:String, 
                    var featureMap: mutable.Map[String, String])




val finalDataset:Dataset[Event] = inputDataFrame.flatMap(
row=> {

        val productId = row.getAs[String]("productId")
        val userActions = row.getAs[Row]("userActions")
        val userEvents:mutable.Seq[Row] = userActions.getAs[mutable.WrappedArray[Row]]("events")

        val processedEvents:mutable.Seq[Row]= userEvents.map(
          event=> 

            val productTypeId = event.getAs[Int]("productTypeId")
            val familyId = event.getAs[String]("familyId")
            val features = activity.getAs[mutable.WrappedArray[Row]]("features")
            val serialId = activity.getAs[String]("serialId")
            val key =  productId ":" serialId
            val features = mutable.Map[String, String]().withDefaultValue(null)
            

            val device_os_list=List("999795_100","987875_101")
            //Feature Map is for every device_os ( example "targetId"->"999795_100") for 999795_100

      if (familyId == 2010 )
    {
      val a: Option[List[String]] = flatten the deviceId,osId ..
      
      a.get.map(i=>{
          val key: String =  methodToCombinedeviceIdAndosId
          val featureMapping: mutable.Map[String, String] = getfeatureMapForInvidualKey

          Event(productId,productTypeId,familyId,key,productName,device_os,feature) ---> This is returning **List[Event]** 
        })
      }
    else{
    Event(productId,productTypeId,familyId,key,productName,device_os,feature)  --> This is returning **Event**. THIS WORKS

    }
            
        )

}

)

uj5u.com熱心網友回復：

我沒有完全一樣地實作它，但我認為有可能理解邏輯并將其應用于您的案例。

我創建了像kafka.json這樣的 json 檔案，并把這樣的代碼放在那里（你的事件）：

[{
  "partition": 1,
  "key": "34768_20220203_MFETP501",
  "offset": 1841543,
  "createTime": 1646041475348,
  "topic": "topic_int",
  "publishTime": 1646041475344,
  "errorCode": 0,
  "userActions": {
    "productId": "3MFETP501",
    "createdDate": "2022-02-26T11:19:35.786Z",
    "events": [
      {
        "GUID": "dbb1-f38b-f7f0-44af-90da-80179412f89c",
        "eventDate": "2022-02-26T11:19:35.786Z",
        "familyId": 2010,
        "productTypeId": 1004678,
        "serialID": "890479804",
        "productName": "MFE Total Protection 2021 Family Pack",
        "features": {
          "mapping": [
            {
              "deviceId": 999795,
              "osId": [
                100
              ]
            },
            {
              "deviceId": 987875,
              "osId": [
                101
              ]
            }
          ]
        }
      }
    ]
  }
}]

請在下面找到基于flatMap和for回圈的第一個解決方案。

  case class Event(
      productId: String,
      familyId: String,
      productTypeId: String,
      key: String,
      productName: String,
      deviceOS: String,
      featureMap: Map[String, String])

  import org.apache.spark.sql.{Dataset, Row, SparkSession}

  import scala.collection.mutable

  val spark = SparkSession
      .builder
      .appName("StructuredStreaming")
      .master("local[*]")
      .getOrCreate()

  private val inputDataFrame = spark.read.option("multiline", "true").format("json").load("/absolute_path_to_kafka.json")


  import spark.implicits._

  val finalDataset: Dataset[Event] = inputDataFrame.flatMap(
    row => {

      val userActions = row.getAs[Row]("userActions")
      val productId = userActions.getAs[String]("productId")

      val userEvents = userActions.getAs[mutable.WrappedArray[Row]]("events")
      for (event <- userEvents;
           familyId = event.getAs[Int]("familyId").toString;
           productTypeId = event.getAs[Int]("productTypeId").toString;
           serialId = event.getAs[String]("serialID");
           productName = event.getAs[String]("productName");
           key = s"$productId:$serialId";
           features = event.getAs[Row]("features");
           mappings = features.getAs[mutable.WrappedArray[Row]]("mapping");
           mappingRow <- mappings;
           deviceId = mappingRow.getAs[Long]("deviceId");
           osIds = mappingRow.getAs[mutable.WrappedArray[Long]]("osId");
           osId <- osIds;
           deviseOs = deviceId   "_"   osId
           ) yield Event(productId, familyId, productTypeId, key, productName, deviseOs, Map("target" -> (deviseOs)))
    }

  )

  finalDataset.foreach(e => println(e))

//  Event(3MFETP501,2010,1004678,3MFETP501:890479804,MFE Total Protection 2021 Family Pack,999795_100,Map(target -> 999795_100))
//  Event(3MFETP501,2010,1004678,3MFETP501:890479804,MFE Total Protection 2021 Family Pack,987875_101,Map(target -> 987875_101))

此外，您可以使用select、withColumn、explode、concat函式來解決此任務。

  case class Event(
      productId: String,
      familyId: String,
      productTypeId: String,
      key: String,
      productName: String,
      deviceOS: String,
      featureMap: Map[String, String])

  import org.apache.spark.sql.{Dataset, SparkSession}
  import org.apache.spark.sql.functions.{col, explode, concat, lit, map}

  val spark = SparkSession
      .builder
      .appName("StructuredStreaming")
      .master("local[*]")
      .getOrCreate()

  private val inputDataFrame = spark.read.option("multiline", "true").format("json").load("/absolute_path_to_kafka.json")

  val transformedDataFrame = inputDataFrame
      .select(col("userActions.productId").as("productId"),
        explode(col("userActions.events")).as("event"))
      .select(col("productId"),
        col("event.familyId").as("familyId"),
        col("event.productTypeId").as("productTypeId"),
        col("event.serialID").as("serialID"),
        col("event.productName").as("productName"),
        explode(col("event.features.mapping")).as("features")
      )
      .select(
        col("productId"),
        col("familyId"),
        col("productTypeId"),
        col("serialID"),
        col("productName"),
        col("features.deviceId").as("deviceId"),
        explode(col("features.osId")).as("osId")
      )
      .withColumn("key", concat(col("productId"), lit(":"), col("serialID")))
      .withColumn("deviceOS", concat(col("deviceId"), lit("_"), col("osId")))
      .withColumn("featureMap", map(lit("target"), col("deviceOS")))

  import spark.implicits._

  private val result: Dataset[Event] = transformedDataFrame.as[Event]
  result.foreach(e => println(e))

//  Event(3MFETP501,2010,1004678,3MFETP501:890479804,MFE Total Protection 2021 Family Pack,999795_100,Map(target -> 999795_100))
//  Event(3MFETP501,2010,1004678,3MFETP501:890479804,MFE Total Protection 2021 Family Pack,987875_101,Map(target -> 987875_101))

添加選項以根據欄位的值之一自定義回應。我在這里替換為對map/flatmap的理解，因此您可以根據型別回傳一個或多個事件作為回應。此外，我對 json 進行了一些自定義以在結果中顯示更多示例。

新的json：

[{
  "partition": 1,
  "key": "34768_20220203_MFETP501",
  "offset": 1841543,
  "createTime": 1646041475348,
  "topic": "topic_int",
  "publishTime": 1646041475344,
  "errorCode": 0,
  "userActions": {
    "productId": "3MFETP501",
    "createdDate": "2022-02-26T11:19:35.786Z",
    "events": [
      {
        "GUID": "dbb1-f38b-f7f0-44af-90da-80179412f89c",
        "eventDate": "2022-02-26T11:19:35.786Z",
        "familyId": 2010,
        "productTypeId": 1004678,
        "serialID": "890479804",
        "productName": "MFE Total Protection 2021 Family Pack",
        "features": {
          "mapping": [
            {
              "deviceId": 999795,
              "osId": [
                100,
                110
              ]
            },
            {
              "deviceId": 987875,
              "osId": [
                101
              ]
            }
          ]
        }
      },
      {
        "GUID": "1111-2222-f7f0-44af-90da-80179412f89c",
        "eventDate": "2022-03-26T11:19:35.786Z",
        "familyId": 2011,
        "productTypeId": 1004679,
        "serialID": "890479805",
        "productName": "Product name",
        "features": {
          "mapping": [
            {
              "deviceId": 999796,
              "osId": [
                103
              ]
            },
            {
              "deviceId": 987877,
              "osId": [
                104
              ]
            }
          ]
        }
      }
    ]
  }
}]

請在下面找到代碼：

  case class Event(
      productId: String,
      familyId: String,
      productTypeId: String,
      key: String,
      productName: String,
      deviceOS: String,
      featureMap: Map[String, String])

  import org.apache.spark.sql.{Dataset, SparkSession}

  val spark = SparkSession
      .builder
      .appName("StructuredStreaming")
      .master("local[*]")
      .getOrCreate()

  private val inputDataFrame = spark.read.option("multiline", "true").format("json").load("/absolute_path_to_kafka.json")

  import spark.implicits._

  val finalDataset: Dataset[Event] = inputDataFrame.flatMap(
    row => {

      val userActions = row.getAs[Row]("userActions")
      val productId = userActions.getAs[String]("productId")

      val userEvents = userActions.getAs[mutable.WrappedArray[Row]]("events")
      for (event <- userEvents;
           productTypeId = event.getAs[Int]("productTypeId").toString;
           serialId = event.getAs[String]("serialID");
           productName = event.getAs[String]("productName");
           key = s"$productId:$serialId";
           familyId = event.getAs[Int]("familyId").toString;
           features = event.getAs[Row]("features");
           mappings = features.getAs[mutable.WrappedArray[Row]]("mapping");
           mappingRow <- mappings;
           deviceId = mappingRow.getAs[Long]("deviceId");
           osIds = mappingRow.getAs[mutable.WrappedArray[Long]]("osId");
           osId <- osIds;
           deviseOs = deviceId   "_"   osId
           ) yield Event(productId, familyId, productTypeId, key, productName, deviseOs, Map("target" -> deviseOs))

      userEvents.flatMap(event => {
        val productTypeId = event.getAs[Int]("productTypeId").toString
        val serialId = event.getAs[String]("serialID")
        val productName = event.getAs[String]("productName")
        val key = s"$productId:$serialId"
        val familyId = event.getAs[Long]("familyId")

        if(familyId == 2010) {
          val features = event.getAs[Row]("features")
          val mappings = features.getAs[mutable.WrappedArray[Row]]("mapping")
          mappings.flatMap(mappingRow => {
            val deviceId = mappingRow.getAs[Long]("deviceId")
            val osIds = mappingRow.getAs[mutable.WrappedArray[Long]]("osId")
            osIds.map(osId => {
              val devise_os = deviceId   "_"   osId
              Event(productId, familyId.toString, productTypeId, key, productName, devise_os, Map("target" -> devise_os))
            })
          })
        } else {
          Seq(Event(productId, familyId.toString, productTypeId, key, productName, "default_defice_os", Map("target" -> "default_defice_os")))
        }
      })

    }
  )

  finalDataset.foreach(e => println(e))

//  Event(3MFETP501,2010,1004678,3MFETP501:890479804,MFE Total Protection 2021 Family Pack,999795_100,Map(target -> 999795_100))
//  Event(3MFETP501,2010,1004678,3MFETP501:890479804,MFE Total Protection 2021 Family Pack,999795_110,Map(target -> 999795_110))
//  Event(3MFETP501,2010,1004678,3MFETP501:890479804,MFE Total Protection 2021 Family Pack,987875_101,Map(target -> 987875_101))
//  Event(3MFETP501,2011,1004679,3MFETP501:890479805,Product name,default_defice_os,Map(target -> default_defice_os))

uj5u.com熱心網友回復：

由于這是在 DataFrame 行下，回傳 Event 案例類，轉換為 DataSet。這里的問題是針對一個條件，我得到List[Event]和休息型別，我只得到Event類

僅供參考：這不是答案。但我進一步嘗試解決。

if (familyId == 2010 )
    {
      val a: Option[List[String]] = flatten the deviceId,osId ..
      
      a.get.map(i=>{
          val key: String =  methodToCombinedeviceIdAndosId
          val featureMapping: mutable.Map[String, String] = getfeatureMapForInvidualKey

          Event(productId,productTypeId,familyId,key,productName,device_os,feature) ---> This is returning List[Event]
        })
      }
    else{
    Event(productId,productTypeId,familyId,key,productName,device_os,feature)  --> This is returning Event

    }

轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/440232.html

標籤：斯卡拉 apache-spark-数据集

上一篇：AkkaClassicAsk模式。它如何匹配詢問與回應？

下一篇：在scala.js中，“錯誤：找不到模塊'jsdom'”的可能原因是什么？