T
Хммм ситуация когда внутри файла 100 записей и 50 запись битая :( хочется что нибудь чтобы достать хотя бы 49 записей ... в идеале 99 :)
c spark.sql.files.ignoreCorruptFiles=true должен читать фаил частично
Size: a a a
2021 February 04
T
если мы о спраке говорим
S
ну в крайнем случае через спарк прикручу, сейчас просто возникла проблема что файлы побились пришлось из архива выгружать много данных и создавать дубли в таргете :( хотелось бы починить файлы Спасибо
S
Думал уже сам сесть написать, но времени не было :(
T
Только вы потестируйте локально прежде чем спарк тащить я то я могу и ошибаться =)
S
да спарк есть :) просто не моей команде :)
A
Просто вы не сказали ни формат данных ни чем читать хотите :)
A
Модет это что кастомные и можно поправить код
A
Ведь если есть код, то дальше можно крутить хоть фильтрацию по полям :)
S
Ну в моем случае это NiFi :) да там можно пдкрутить ридер чтобы он умел битые читать, кстати хорошая мысль сделать авроридер который сможет и битые распарсить и тогда смогу решать на ходу проблему
С
Да, мое описание довольно таки хаотично, вдобавок моими толстыми пальцами трудно попасть в эти маленькие буковки на телефоне 😁
Идея использования delta lake, Hudi, Iceberg пришла от необходимости периодически делать backfill. Насколько я понял, эти технологии могут помочь облегчить эти операции. SCD1 когда-то удовлетворял потребности этой компании, когда данных было немного. Теперь дата сет вырос в несколько раз, железо ненадежное, и нарастить его тоже проблематично и долго. Поэтому я решил перевести всё в режим SCD2, или IDL + deltas.
Идея использования delta lake, Hudi, Iceberg пришла от необходимости периодически делать backfill. Насколько я понял, эти технологии могут помочь облегчить эти операции. SCD1 когда-то удовлетворял потребности этой компании, когда данных было немного. Теперь дата сет вырос в несколько раз, железо ненадежное, и нарастить его тоже проблематично и долго. Поэтому я решил перевести всё в режим SCD2, или IDL + deltas.
Без backfill можно держать две таблицы - одна актуальная, другая с историей, места жрёт немного больше, зато немного экономии на том, что не нужно елозить и филить данные
Table_name
Table_name_log
Table_name
Table_name_log
AE
кто знает, есть ли какой способ выдернуть из SparkHadoopWriter кол-во записей, которые он записал?
AE
у него же вроде как есть метрики, как до них добраться из спарк контекста?
AE
окей, сам спросил, сам нашёл
sc.addSparkListener(new SparkListener() {
override def onTaskEnd(taskEnd: SparkListenerTaskEnd) {
типа так ---> taskEnd.taskMetrics.outputMetrics.recordsWritten
}
})
sc.addSparkListener(new SparkListener() {
override def onTaskEnd(taskEnd: SparkListenerTaskEnd) {
типа так ---> taskEnd.taskMetrics.outputMetrics.recordsWritten
}
})
AE
но теперь вопрос как понять какую именно рддху писала эта таска
V
Они стоят ещё местами, продавать тоже прекратили
Смотря кому. Ритейлу продали
V
Хотя оч активно муссировались слухи, что все, bda почил
ИК
окей, сам спросил, сам нашёл
sc.addSparkListener(new SparkListener() {
override def onTaskEnd(taskEnd: SparkListenerTaskEnd) {
типа так ---> taskEnd.taskMetrics.outputMetrics.recordsWritten
}
})
sc.addSparkListener(new SparkListener() {
override def onTaskEnd(taskEnd: SparkListenerTaskEnd) {
типа так ---> taskEnd.taskMetrics.outputMetrics.recordsWritten
}
})
В полях SparkListenerTaskEnd.TaskInfo есть taskId, оно должно появиться в имени файла. Регексп "part-(\d+)-.*". Часть \d+ - это должно быть taskId
И, если не затруднит, после нескольких миллиардов записей, дайте фидбек, правильно ли считает, Хочу использовать, чтобы не делать запрос с input_file_name()
И, если не затруднит, после нескольких миллиардов записей, дайте фидбек, правильно ли считает, Хочу использовать, чтобы не делать запрос с input_file_name()
А
В полях SparkListenerTaskEnd.TaskInfo есть taskId, оно должно появиться в имени файла. Регексп "part-(\d+)-.*". Часть \d+ - это должно быть taskId
И, если не затруднит, после нескольких миллиардов записей, дайте фидбек, правильно ли считает, Хочу использовать, чтобы не делать запрос с input_file_name()
И, если не затруднит, после нескольких миллиардов записей, дайте фидбек, правильно ли считает, Хочу использовать, чтобы не делать запрос с input_file_name()
Если таска рестартовалась, то говорят счётчик не сбрасывается
ИК
Если таска рестартовалась, то говорят счётчик не сбрасывается
вот и я об этом беспокоюсь