G
Хорошо. так а какое постоянное решение если нужна сквазная нумерация ID
Size: a a a
2020 October 20
AS
Хорошо. так а какое постоянное решение если нужна сквазная нумерация ID
zipwithindex как уже подсказали, будет шафл, но всяко лучше чем window
G
row_number плохо - потому что медленно и шафл. monotonically_incrace_id - вообще не подходит так как генериться будут ID ghb rf;ljv pfgecrt
G
zipwithindex как уже подсказали, будет шафл, но всяко лучше чем window
Понял
t
Хорошо. так а какое постоянное решение если нужна сквазная нумерация ID
никак
t
не делать такого в спарке :)
AS
не делать такого в спарке :)
лучше быть богатым и здоровым
G
Ну лучше ничего не делать. Так и не ошибиться.
R
можно попробовать хэш натравить на уникальный набор полей
NG
Коллеги, привет!
Хотел посоветоваться на счет graceful shutdown. Есть стрим kafka->hdfs с несложной логикой внутри.
Если сделать yarn -kill, есть высокий шанс получить дубли в приемнике при рестарте. Нагуглил путь с подкладыванием файлика и остановкой стрима при его удалении. Какие есть еще варианты? Смотрел в сторону
Spark 2.4.5
Хотел посоветоваться на счет graceful shutdown. Есть стрим kafka->hdfs с несложной логикой внутри.
Если сделать yarn -kill, есть высокий шанс получить дубли в приемнике при рестарте. Нагуглил путь с подкладыванием файлика и остановкой стрима при его удалении. Какие есть еще варианты? Смотрел в сторону
spark.streaming.stopGracefullyOnShutdown, но так понял, это не очень живая вещь. В общем, в поисках best practice =)Spark 2.4.5
t
ты пишешь writeStream в хдфс или foreachBatch и внутри пишешь в хдфс?
NG
ты пишешь writeStream в хдфс или foreachBatch и внутри пишешь в хдфс?
forEachBatch
t
тогда в начале функции кидай исключение когда тебе надо остановить твой джоб и отавливай его в sq.awaitTermination
NG
тогда в начале функции кидай исключение когда тебе надо остановить твой джоб и отавливай его в sq.awaitTermination
Спасибо! буду пробовать
AS
Переслано от Denis Tsvetkov
что-нибудь вроде
import sun.misc.{Signal, SignalHandler}
var isTimeToStop = false
// в самом начале регистрируем обработчик
registerSignalHandlers()
def registerSignalHandlers() = {
val signalHandler = new SignalHandler {
override def handle(signal: Signal): Unit = {
log.info(s"Signal ${signal} received.")
isTimeToStop = true
}
Signal.handle(new Signal("TERM"), signalHandler)
}ну и перед обработкой каждого из батчей проверять, не пришло ли время isTimeToStop == true ?2020 October 21
K
Переслано от Denis Tsvetkov
что-нибудь вроде
import sun.misc.{Signal, SignalHandler}
var isTimeToStop = false
// в самом начале регистрируем обработчик
registerSignalHandlers()
def registerSignalHandlers() = {
val signalHandler = new SignalHandler {
override def handle(signal: Signal): Unit = {
log.info(s"Signal ${signal} received.")
isTimeToStop = true
}
Signal.handle(new Signal("TERM"), signalHandler)
}ну и перед обработкой каждого из батчей проверять, не пришло ли время isTimeToStop == true ?в ярне работать не будет
VM
Коллеги, а как в structured стриминге метрики принято собирать?
K
статсд