PK
Size: a a a
2018 December 14
t
https://github.com/tenkeiu8/spark-streaming-jdbc < пример кастомного сурса в продолжение митапа
AS
https://github.com/tenkeiu8/spark-streaming-jdbc < пример кастомного сурса в продолжение митапа
👍
GP
https://github.com/tenkeiu8/spark-streaming-jdbc < пример кастомного сурса в продолжение митапа
Делай Ридмиху и мы зафорсим
2018 December 15
Ds
Ребят, спасибо за митап! Все было на высоте.
PK
Да, всем спасибо, был приятно удивлён количеству народа перед НГ. Презентации и видео выложу в пнд-вт
AD
https://github.com/tenkeiu8/spark-streaming-jdbc < пример кастомного сурса в продолжение митапа
Спасибо, а может кто еще кинуть ссылочку на PR Structured streaming listener, который упоминался в вопросах в обсуждении данного кастомного сурса?
AS
Спасибо, а может кто еще кинуть ссылочку на PR Structured streaming listener, который упоминался в вопросах в обсуждении данного кастомного сурса?
pr не к листенеру, листенер обычный(https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#reporting-metrics-programmatically-using-asynchronous-apis), но если делать кастомный sink, как онисено например здесь(https://github.com/jaceklaskowski/spark-structured-streaming-book/blob/master/spark-sql-streaming-demo-custom-sink-webui.adoc), то метрики работать не будут.
Вот пример, того что надо сделать в кстомном sink, что бы все заработало
https://github.com/sutugin/shc/blob/master/core/src/main/scala/org/apache/spark/sql/execution/streaming/HBaseStreamSinkProvider.scala#L29
Вот пример, того что надо сделать в кстомном sink, что бы все заработало
https://github.com/sutugin/shc/blob/master/core/src/main/scala/org/apache/spark/sql/execution/streaming/HBaseStreamSinkProvider.scala#L29
t
pr не к листенеру, листенер обычный(https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#reporting-metrics-programmatically-using-asynchronous-apis), но если делать кастомный sink, как онисено например здесь(https://github.com/jaceklaskowski/spark-structured-streaming-book/blob/master/spark-sql-streaming-demo-custom-sink-webui.adoc), то метрики работать не будут.
Вот пример, того что надо сделать в кстомном sink, что бы все заработало
https://github.com/sutugin/shc/blob/master/core/src/main/scala/org/apache/spark/sql/execution/streaming/HBaseStreamSinkProvider.scala#L29
Вот пример, того что надо сделать в кстомном sink, что бы все заработало
https://github.com/sutugin/shc/blob/master/core/src/main/scala/org/apache/spark/sql/execution/streaming/HBaseStreamSinkProvider.scala#L29
что примечательно, df.rdd также делает queryExecution.toRdd
t
lazy val rdd: RDD[T] = {
val objectType = exprEnc.deserializer.dataType
rddQueryExecution.toRdd.mapPartitions { rows =>
rows.map(_.get(0, objectType).asInstanceOf[T])
}
}
val objectType = exprEnc.deserializer.dataType
rddQueryExecution.toRdd.mapPartitions { rows =>
rows.map(_.get(0, objectType).asInstanceOf[T])
}
}
t
и на нем вотермарки не работают
t
а вот если в синке юзать data.queryExecution.toRdd, то все ок
t
t
а, ну походу из-за этого:
@transient private lazy val rddQueryExecution: QueryExecution = {
val deserialized = CatalystSerde.deserialize[T](planWithBarrier)
sparkSession.sessionState.executePlan(deserialized)
}
@transient private lazy val rddQueryExecution: QueryExecution = {
val deserialized = CatalystSerde.deserialize[T](planWithBarrier)
sparkSession.sessionState.executePlan(deserialized)
}
AD
pr не к листенеру, листенер обычный(https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#reporting-metrics-programmatically-using-asynchronous-apis), но если делать кастомный sink, как онисено например здесь(https://github.com/jaceklaskowski/spark-structured-streaming-book/blob/master/spark-sql-streaming-demo-custom-sink-webui.adoc), то метрики работать не будут.
Вот пример, того что надо сделать в кстомном sink, что бы все заработало
https://github.com/sutugin/shc/blob/master/core/src/main/scala/org/apache/spark/sql/execution/streaming/HBaseStreamSinkProvider.scala#L29
Вот пример, того что надо сделать в кстомном sink, что бы все заработало
https://github.com/sutugin/shc/blob/master/core/src/main/scala/org/apache/spark/sql/execution/streaming/HBaseStreamSinkProvider.scala#L29
спасибо за подробный ответ!
2018 December 16
AD
@tenKe @sutuginandrey
В Spark 2.4.0 проблемму кастомных синков для сохранения где либо решили штатно: https://stackoverflow.com/a/51319121
В Spark 2.4.0 проблемму кастомных синков для сохранения где либо решили штатно: https://stackoverflow.com/a/51319121
AS
@tenKe @sutuginandrey
В Spark 2.4.0 проблемму кастомных синков для сохранения где либо решили штатно: https://stackoverflow.com/a/51319121
В Spark 2.4.0 проблемму кастомных синков для сохранения где либо решили штатно: https://stackoverflow.com/a/51319121
не очень понял как это связано с кастомными синками... там вроде обсуждается проблема записи во множество синков...
t
@tenKe @sutuginandrey
В Spark 2.4.0 проблемму кастомных синков для сохранения где либо решили штатно: https://stackoverflow.com/a/51319121
В Spark 2.4.0 проблемму кастомных синков для сохранения где либо решили штатно: https://stackoverflow.com/a/51319121
тут не совсем вопрос кастмомных синков решается, а скорее как на наделать кучу выходных стримов
AD
не очень понял как это связано с кастомными синками... там вроде обсуждается проблема записи во множество синков...
Имел ввиду, что теперь это можно использовать вместо
addBatch https://github.com/sutugin/shc/blob/master/core/src/main/scala/org/apache/spark/sql/execution/streaming/HBaseStreamSinkProvider.scala#L29
Написав:
streamingDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
batchDF.write.format("org.apache.spark.sql.execution.datasources.hbase").save(...)
}
addBatch https://github.com/sutugin/shc/blob/master/core/src/main/scala/org/apache/spark/sql/execution/streaming/HBaseStreamSinkProvider.scala#L29
Написав:
streamingDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
batchDF.write.format("org.apache.spark.sql.execution.datasources.hbase").save(...)
}
PK
Имел ввиду, что теперь это можно использовать вместо
addBatch https://github.com/sutugin/shc/blob/master/core/src/main/scala/org/apache/spark/sql/execution/streaming/HBaseStreamSinkProvider.scala#L29
Написав:
streamingDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
batchDF.write.format("org.apache.spark.sql.execution.datasources.hbase").save(...)
}
addBatch https://github.com/sutugin/shc/blob/master/core/src/main/scala/org/apache/spark/sql/execution/streaming/HBaseStreamSinkProvider.scala#L29
Написав:
streamingDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
batchDF.write.format("org.apache.spark.sql.execution.datasources.hbase").save(...)
}
Да, я как раз этот пример приводил в хайлайтах 2.4.0