Н
сорян за такой поток мыслей))
{"text":"large html"}
Вот хочу отпроцессить (найти различные вхождения в html) его через спарк и записать в таблицу типа
domain is_shop ... has_phone_number
google.ru True True
Size: a a a
2020 January 14
Н
каждая строка большой html странички от краулера
{"text":"large html"}
Вот хочу отпроцессить (найти различные вхождения в html) его через спарк и записать в таблицу типа
{"text":"large html"}
Вот хочу отпроцессить (найти различные вхождения в html) его через спарк и записать в таблицу типа
R
привет, у меня flume стримит в папку на hdfs, хочу отпроцессить эти данные и затирать то, что прочитал и писать в hive красиво, чтобы не плодить маленькие файлы.
вопрос: как читать файлы в sparke а потом их затирать? Как лучше делать это джобом или спарк стримингом?
Если писать малеьникими rdd через спарк стриминг в хайв, он их потом замерджит или будут маленькьие файлы?
можешь делать следующее:
- читаешь файловый поток через StructuredStreaming
- После каждого батча берешь имена прочитанных файлов и удаляешь их в исходнике
- Данные пишешь в
- читаешь файловый поток через StructuredStreaming
- После каждого батча берешь имена прочитанных файлов и удаляешь их в исходнике
- Данные пишешь в
delta -формат, время от времени прогоняешь compaction по записанным даннымН
а как узнать имена прочитаных файлов?
R
в scala как минимум есть функция:
import org.apache.spark.sql.functions.input_file_name
R
она вернет для каждой записи имя файла. можно после батча сделать что-то вроде (псевдокод):
def foreachBatch(some_args): Unit = {
val df = processBatch(batch)
val fnames = df.withColumn(“ifn”, input_file_name).select(“input_file_name”).distinct().collect()
fnames.map(deleteFile)
}R
Но вообще обычно стараются не удалять исходники (так называемый raw data level) насовсем. Что будет, если твой процессинг радикально поменялся и тебе нужно сделать refeed и заново отчитать все исходные текстовые файлы?
Н
ну я могу еще раз прогнать, я просто не хочу процесить старые данные, а только новые поступающие
Н
Правильно ли я понимаю, если я перезапущу скрипт со structured streaming, то он начнет работать с самого начала
R
ну я могу еще раз прогнать, я просто не хочу процесить старые данные, а только новые поступающие
так зачем файлы то удалять после прочтения тогда?
https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#input-sources
спарк будет автоматически помечать прочитанные файлы, но нужно будет включить checkpointing. Чекпоинт будет хранить информацию о последнем прочитанном файле
https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#input-sources
спарк будет автоматически помечать прочитанные файлы, но нужно будет включить checkpointing. Чекпоинт будет хранить информацию о последнем прочитанном файле
Н
о, хорошо тогда
R
вот тут про чекпоинт можно почитать:
https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#recovering-from-failures-with-checkpointing
https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#recovering-from-failures-with-checkpointing
Н
спс
2020 January 15
V
Привет, при сабмите через Livy нескольких spark апликейшенов( около 10), несколько из них падает с ошибкой http 409 конфликт при записи. Может кто-нибудь сталкивался с подобным?
A
а какая версия livy?
у меня правда ошибки такой не было и в коде livy про конфликт тоже ничего нету (по крайней мере в версии 0.5.х)
у меня правда ошибки такой не было и в коде livy про конфликт тоже ничего нету (по крайней мере в версии 0.5.х)
V
EV
Всем приветик,
есть задачка обстреливать HTTP API из Flink DataStream, как правильно это реализовать если нужна гарантия доставки at least once? У меня варианты это RichAsyncFunction или кастомный синк
Я пока не понял как это все поведет себя в случае отказа downstream API
есть задачка обстреливать HTTP API из Flink DataStream, как правильно это реализовать если нужна гарантия доставки at least once? У меня варианты это RichAsyncFunction или кастомный синк
Я пока не понял как это все поведет себя в случае отказа downstream API
EP
Коллеги, такой вопрос, как мне в спарке выполнить единоразовую операцию на каждом экзекьюторе? Например зарегистрировать jdbc драйвер
EV
Коллеги, такой вопрос, как мне в спарке выполнить единоразовую операцию на каждом экзекьюторе? Например зарегистрировать jdbc драйвер
mapPartitions
EP
Да хотел уточнить, что там стриминг и mapPartitions будет регистрировать драйвер на на каждом батче. Или просто использовать флаг, что драйвер уже зарегистрирован?