我有一個 spark-streaming 應用程式，我想在我的主要操作之前進行一些資料轉換，但轉換涉及一些資料驗證。
當驗證失敗時，我想記錄失敗案例，然后繼續其余的。
目前，我有這樣的代碼：
def values: DStream[String] = ???
def validate(element: String): Either[String, MyCaseClass] = ???
val validationResults = values.map(validate)
validationResults.foreachRDD { rdd =>
rdd.foreach {
case Left(error) => logger.error(error)
case _ =>
}
}
val validatedValues: DStream[MyCaseClass] =
validationResults.mapPartitions { partition =>
partition.collect { case Right(record) => record }
}
這目前有效，但感覺就像我做錯了什么。
問題
據我了解，這將執行validation兩次，這可能是浪費。
values.map(validation).persist()用于解決該問題是否正確？- 即使我堅持這些值，它仍然會在所有元素上迭代和模式匹配兩次。感覺應該有一些方法可以用來解決這個問題。在常規的 scala 集合中，我可能會使用一些貓
TraverseFilterapi，或者fs2.Stream使用evalMapFilter. 我可以為此使用什么 DStream api？也許有什么mapPartitions？
uj5u.com熱心網友回復：
我想說解決這個問題的最好方法是利用 stdlibflatMap接受的優勢Option
def values: DStream[String] = ???
def validate(element: String): Either[String, MyCaseClass] = ???
val validatedValues: DStream[MyCaseClass] =
values.map(validate).flatMap {
case Left(error) =>
logger.error(error)
None
case Right(record) =>
Some(record)
}
您也可以使用更詳細一點mapPartitions，這應該更有效。
uj5u.com熱心網友回復：
這里的“最佳”選項在一定程度上取決于您的 spark 作業的其余部分和您的 spark 版本。
理想情況下，您會選擇催化劑本身可以理解的機制。spark3 資料集觀察偵聽器可能是您正在尋找的。（我還沒有看到很多在野外使用它的例子，但似乎這就是這件事背后的動機。）
在純 spark sql 中，一種選擇是添加一個帶有驗證結果的新列，例如invalid_reason，NULL如果記錄有效，則命名為列，或者包含列驗證失敗原因的一些 [列舉] 字串。此時，您可能希望在執行groupBy/count/collect/log操作之前持久化/快取資料集，然后過濾where invalid_reason is null持久化的資料幀并繼續進行其余處理。
tl; dr：考慮添加驗證列，而不僅僅是應用“驗證”功能。然后，您在此處“分叉”處理：記錄指定了無效列的記錄，處理未指定的記錄的其余作業。它確實為您的資料框增加了一些容量，但不需要兩次處理相同的記錄。
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/441152.html