Блог

Конвейер слияния данных выдает нам один или несколько файлов деталей на выходе, если синхронизировать в корзине GCS. Мой вопрос в том, как мы можем объединить эти файлы деталей в один, а также присвоить им значимое имя?

Data Fusion Преобразования выполняются в Dataproc кластерах, выполняющих либо Spark , либо MapReduce задания. Ваш конечный результат разделен на множество файлов, поскольку задания разделяют ваши данные на основе разделов HDFS (это поведение по умолчанию для Spark / Hadoop).

При написании сценария Spark вы можете управлять этим поведением по умолчанию и создавать разные выходные данные. Однако Data Fusion был создан для абстрагирования уровня кода и предоставления вам опыта использования полностью управляемого интегратора данных. Использование разделенных файлов не должно быть проблемой, но если вам действительно нужно их объединить, я предлагаю вам использовать следующий подход:

1. В верхней части вашей Pipeline Studio нажмите на Концентратор -> Плагины, найдите Динамический плагин Spark, нажмите на Развертывание, а затем на Готово (вы можете игнорировать файл JAR)

2. Вернитесь к своему конвейеру, выберите Spark в разделе приемника.

3. Замените свой плагин GCS на плагин Spark

4. В вашем плагине Spark установите для компиляции во время развертывания значение false и замените код некоторым кодом Spark, который выполняет то, что вы хотите. Приведенный ниже код, например, жестко запрограммирован, но работает:

     def sink(df: DataFrame) : Unit = {
      new_df = df.coalesce(1)
      new_df.write.format("csv").save("gs://your/path/")
    }
     

   Эта функция получает данные из вашего конвейера в виде фрейма данных. Функция coalesce уменьшает количество разделов до 1, а последняя строка записывает его в GCS.

5. Разверните свой конвейер, и он будет готов к запуску