Прием данных CSV в BigQuery с помощью Cloud Data Fusion – пакетный прием

1. Введение

Последнее обновление: 28 февраля 2020 г.

В этой кодовой лаборатории демонстрируется шаблон приема данных для массовой загрузки медицинских данных в формате CSV в BigQuery. Для этой лабораторной работы мы будем использовать конвейер пакетных данных Cloud Data Fusion. Реалистичные данные медицинских тестов были созданы и доступны для вас в сегменте Google Cloud Storage ( gs://hcls_testing_data_fhir_10_phases/csv/ ).

В этой лабораторной работе по кодированию вы узнаете:

• Как импортировать данные CSV (пакетная загрузка по расписанию) из GCS в BigQuery с помощью Cloud Data Fusion .
• Как визуально построить конвейер интеграции данных в Cloud Data Fusion для массовой загрузки, преобразования и маскировки медицинских данных .

Что вам нужно для запуска этой лаборатории кода?

• Вам нужен доступ к проекту GCP.
• Вам должна быть назначена роль владельца проекта GCP.
• Данные о здравоохранении в формате CSV, включая заголовок.

Если у вас нет проекта GCP, выполните следующие действия , чтобы создать новый проект GCP.

Медицинские данные в формате CSV были предварительно загружены в корзину GCS по адресу gs://hcls_testing_data_fhir_10_phases/csv/ . Каждый CSV-файл ресурса имеет свою уникальную структуру схемы. Например, в файле Patients.csv используется другая схема, чем в файле Providers.csv. Предварительно загруженные файлы схем можно найти по адресу gs://hcls_testing_data_fhir_10_phases/csv_schemas .

Если вам понадобится новый набор данных, вы всегда можете сгенерировать его с помощью SyntheaTM . Затем загрузите его в GCS вместо копирования из корзины на этапе копирования входных данных.

Ссылка:

Основные операции в Cloud Data Fusion (краткое руководство ) с использованием консоли Google Cloud Platform.

2. Настройка проекта GCP

Инициализируйте переменные оболочки для вашей среды.

Чтобы найти PROJECT_ID , обратитесь к разделу Идентификация проектов .

<!-- CODELAB: Initialize shell variables ->
<!-- Your current GCP Project ID ->
export PROJECT_ID=<PROJECT_ID>
<!-- A new GCS Bucket in your current Project  - INPUT ->
export BUCKET_NAME=<BUCKET_NAME>
<!-- A new BQ Dataset ID - OUTPUT ->
export DATASET_ID=<DATASET_ID>

Создайте сегмент GCS для хранения входных данных и журналов ошибок с помощью инструмента gsutil .

gsutil mb -l us gs://$BUCKET_NAME

Получите доступ к синтетическому набору данных.

1. С адреса электронной почты, который вы используете для входа в Cloud Console, отправьте электронное письмо на адрес [e-mail] с просьбой присоединиться.
2. Вы получите электронное письмо с инструкциями о том, как подтвердить действие.
3. Используйте возможность ответить на электронное письмо, чтобы присоединиться к группе. НЕ нажимайте кнопку.
4. Как только вы получите электронное письмо с подтверждением, вы сможете перейти к следующему шагу в кодовой лаборатории.

Скопируйте входные данные.

gsutil -m cp -r gs://hcls_testing_data_fhir_10_patients/csv gs://$BUCKET_NAME

Создайте набор данных BigQuery.

bq mk --location=us --dataset $PROJECT_ID:$DATASET_ID

Помните: вы всегда можете создать новый набор реалистичных тестовых данных, используя SyntheaTM для этой лабораторной работы. Для вашего удобства мы предоставляем набор данных в GCS: gs://hcls_testing_data_fhir_10_phases/csv.

3. Настройка среды Cloud Data Fusion

Выполните следующие действия, чтобы включить API Cloud Data Fusion и предоставить необходимые разрешения:

Включите API .

1. Перейдите в библиотеку API консоли GCP .
2. В списке проектов выберите свой проект.
3. В библиотеке API выберите API, который вы хотите включить. Если вам нужна помощь в поиске API, воспользуйтесь полем поиска и/или фильтрами.
4. На странице API нажмите ВКЛЮЧИТЬ.

Создайте экземпляр Cloud Data Fusion .

1. В консоли GCP выберите свой ProjectID.
2. Выберите Data Fusion в левом меню, затем нажмите кнопку СОЗДАТЬ ЭКЗЕМПЛЯР в середине страницы (первое создание) или нажмите кнопку СОЗДАТЬ ЭКЗЕМПЛЯР в верхнем меню (дополнительное создание).

3. Укажите имя экземпляра. Выберите Предприятие.

4. Нажмите кнопку СОЗДАТЬ.

Примечание. Процесс создания экземпляра может занять до 30 минут.

Разрешения экземпляра установки.

После создания экземпляра выполните следующие действия, чтобы предоставить учетной записи службы, связанной с разрешениями экземпляра, в вашем проекте:

Примечание. Предоставление разрешения необходимо выполнить только один раз в проекте.

1. Перейдите на страницу сведений об экземпляре, щелкнув имя экземпляра.

2. Скопируйте учетную запись службы.

3. Перейдите на страницу IAM вашего проекта.
4. На странице разрешений IAM мы теперь добавим сервисный аккаунт в качестве нового участника и предоставим ему роль агента службы Cloud Data Fusion API . Нажмите кнопку «Добавить» , затем вставьте «служебный аккаунт» в поле «Новые участники» и выберите «Управление услугами» -> Роль агента сервера Cloud Data Fusion API.
5. 
6. Нажмите Сохранить .

После выполнения этих шагов вы можете начать использовать Cloud Data Fusion, щелкнув ссылку «Просмотреть экземпляр» на странице экземпляров Cloud Data Fusion или странице сведений об экземпляре.

Настройте правило брандмауэра.

1. Перейдите в Консоль GCP -> Сеть VPC -> Правила брандмауэра, чтобы проверить, существует ли правило default-allow-ssh или нет.

2. Если нет, добавьте правило брандмауэра, которое разрешает весь входящий SSH-трафик в сеть по умолчанию.

Использование командной строки:

gcloud beta compute --project={PROJECT_ID} firewall-rules create default-allow-ssh --direction=INGRESS --priority=1000 --network=default --action=ALLOW --rules=tcp:22 --source-ranges=0.0.0.0/0 --enable-logging

Использование пользовательского интерфейса: нажмите «Создать правило брандмауэра» и заполните информацию:

4. Постройте схему трансформации

Теперь, когда у нас есть среда Cloud Fusion в GCP, давайте построим схему. Эта схема нам нужна для преобразования данных CSV.

1. В окне Cloud Data Fusion нажмите ссылку «Просмотреть экземпляр» в столбце «Действие». Вы будете перенаправлены на другую страницу. Нажмите предоставленный URL-адрес , чтобы открыть экземпляр Cloud Data Fusion. По вашему выбору нажмите кнопку «Начать тур» или кнопку «Нет, спасибо» во всплывающем окне приветствия.
2. Разверните меню «гамбургер», выберите Pipeline -> Studio.

3. В разделе «Преобразование» на палитре плагинов слева дважды щелкните узел Wrangler, который появится в пользовательском интерфейсе конвейеров данных.

4. Наведите указатель мыши на узел Wrangler и нажмите «Свойства» . Нажмите кнопку «Сборка» , затем выберите исходный файл .csv (например, Patients.csv), который должен содержать все поля данных для построения нужной схемы.
5. Нажмите стрелку вниз («Преобразования столбцов») рядом с именем каждого столбца (например, «тело»).
6. По умолчанию при первоначальном импорте предполагается, что в вашем файле данных есть только один столбец. Чтобы проанализировать его как CSV, выберите «Разбор → CSV» , затем выберите разделитель и установите соответствующий флажок «Установить первую строку как заголовок». Нажмите кнопку «Применить».
7. Нажмите стрелку вниз рядом с полем «Тело» и выберите «Удалить столбец», чтобы удалить поле «Тело». Кроме того, вы можете попробовать другие преобразования, такие как удаление столбцов, изменение типа данных для некоторых столбцов (по умолчанию используется тип «строка»), разделение столбцов, установка имен столбцов и т. д.

8. На вкладках «Столбцы» и «Шаги преобразования» показана схема вывода и рецепт Wrangler. Нажмите Применить в правом верхнем углу. Нажмите кнопку «Подтвердить». Зеленый цвет «Ошибок не обнаружено» указывает на успех.

9. В свойствах Wrangler щелкните раскрывающийся список «Действия» , чтобы экспортировать нужную схему в локальное хранилище для будущего импорта , если это необходимо.
10. Сохраните рецепт Wrangler для дальнейшего использования.

parse-as-csv :body ',' true
drop body

11. Чтобы закрыть окно свойств Wrangler, нажмите кнопку X.

5. Создайте узлы для конвейера

В этом разделе мы построим компоненты конвейера.

1. В пользовательском интерфейсе конвейеров данных в левом верхнем углу вы должны увидеть, что в качестве типа конвейера выбран «Конвейер данных — пакет» .

2. На левой панели есть различные разделы: «Фильтр», «Источник», «Преобразование», «Аналитика», «Приемник», «Условия и действия», «Обработчики ошибок» и «Предупреждения», где вы можете выбрать узел или узлы для конвейера.

Исходный узел

1. Выберите узел Источник.
2. В разделе «Источник» на палитре плагинов слева дважды щелкните узел Google Cloud Storage , который отображается в пользовательском интерфейсе конвейеров данных.
3. Укажите исходный узел GCS и нажмите «Свойства» .

4. Заполните необходимые поля. Установите следующие поля:

• Метка = {любой текст}
• Имя ссылки = {любой текст}
• Идентификатор проекта = автоматическое определение
• Путь = URL-адрес GCS для сегмента в текущем проекте. Например, gs://$BUCKET_NAME/csv/
• Формат = текст
• Поле пути = имя файла
• Только имя файла пути = true
• Читать файлы рекурсивно = true

5. Добавьте поле «имя файла» в схему вывода GCS, нажав кнопку «+» .
6. Нажмите «Документация» для подробного объяснения. Нажмите кнопку «Подтвердить». Зеленый цвет «Ошибок не обнаружено» указывает на успех.
7. Чтобы закрыть свойства GCS, нажмите кнопку X.

Узел преобразования

1. Выберите узел Преобразование.
2. В разделе «Преобразование» на палитре плагинов слева дважды щелкните узел Wrangler , который появляется в пользовательском интерфейсе конвейеров данных. Подключите исходный узел GCS к узлу преобразования Wrangler.
3. Наведите указатель мыши на узел Wrangler и нажмите «Свойства» .
4. Нажмите раскрывающийся список «Действия» и выберите «Импорт» , чтобы импортировать сохраненную схему (например: gs://hcls_testing_data_fhir_10_phases/csv_schemas/ schema (Patients).json ), и вставьте сохраненный рецепт из предыдущего раздела.
5. Или повторно используйте узел Wrangler из раздела: Построение схемы для трансформации .
6. Заполните необходимые поля. Установите следующие поля:

• Метка = {любой текст}
• Имя поля ввода = {*}
• Предварительное условие = {имя_файла != "пациенты.csv"}, чтобы отличать каждый входной файл ( например, Patients.csv, поставщики.csv, аллергии.csv и т. д. ) от исходного узла.

7. Добавьте узел JavaScript для выполнения предоставленного пользователем кода JavaScript, который дополнительно преобразует записи. В этой кодовой лаборатории мы используем узел JavaScript для получения отметки времени для каждого обновления записи. Подключите узел преобразования Wrangler к узлу преобразования JavaScript. Откройте « Свойства JavaScript» и добавьте следующую функцию:

function transform(input, emitter, context) {
  input.TIMESTAMP = (new Date()).getTime()*1000;
  emitter.emit(input);
}

8. Добавьте поле с именем TIMESTAMP в схему вывода (если оно не существует), щелкнув знак + . Выберите временную метку в качестве типа данных.

9. Нажмите «Документация» для получения подробного объяснения. Нажмите кнопку «Проверить», чтобы проверить всю входную информацию. Зеленый цвет «Ошибок не обнаружено» указывает на успех.
10. Чтобы закрыть окно «Свойства преобразования», нажмите кнопку X.

Маскирование и деидентификация данных

1. Вы можете выбрать отдельные столбцы данных, щелкнув стрелку вниз в столбце и применив правила маскирования в разделе «Выбор данных маски» в соответствии с вашими требованиями (например, столбец SSN).

2. Вы можете добавить больше директив в окне «Рецепт» узла Wrangler . Например, используя директиву hash с алгоритмом хеширования, следующим синтаксисом, в целях деидентификации:

hash <column> <algorithm> <encode>

<column>: name of the column
<algorithm>: Hashing algorithm (i.e. MD5, SHA-1, etc.)
<encode>: default is true (hashed digest is encoded as hex with left-padding zeros). To disable hex encoding, set <encode> to false.

Примечание. Значение True означает анализ столбца тела с установленной опцией «Установить первую строку как заголовок».

Приемный узел

1. Выберите узел приемника.
2. В разделе «Приемник» на палитре плагинов слева дважды щелкните узел BigQuery, который появится в пользовательском интерфейсе конвейера данных.
3. Наведите указатель мыши на узел приемника BigQuery и нажмите «Свойства».

4. Заполните необходимые поля. Установите следующие поля:

• Метка = {любой текст}
• Имя ссылки = {любой текст}
• Идентификатор проекта = автоматическое определение
• Набор данных = набор данных BigQuery, используемый в текущем проекте (т. е. DATASET_ID).
• Таблица = {имя таблицы}

5. Нажмите «Документация» для получения подробного объяснения. Нажмите кнопку «Проверить», чтобы проверить всю входную информацию. Зеленый цвет «Ошибок не обнаружено» указывает на успех.

6. Чтобы закрыть свойства BigQuery, нажмите кнопку X.

6. Создайте конвейер пакетных данных

Соединение всех узлов в конвейере

1. Перетащите стрелку соединения > на правый край исходного узла и опустите на левый край узла назначения.
2. Конвейер может иметь несколько ветвей, которые получают входные файлы из одного и того же узла источника GCS.

Примечание. Создайте одну ветвь исходного узла в конвейере для каждого входного файла .csv.

3. Назовите трубопровод.

Примечание. На протяжении всего этого процесса схема автоматически распространяется по мере построения конвейера. Вы также можете импортировать нужную схему.

Вот и все. Вы только что создали свой первый конвейер пакетных данных и можете его развернуть и запустить.

Отправка оповещений конвейера по электронной почте (необязательно)

Чтобы использовать функцию Pipeline Alert SendEmail, конфигурация требует, чтобы почтовый сервер был настроен для отправки почты с экземпляра виртуальной машины. Для получения дополнительной информации см. ссылку ниже:

Отправка электронной почты из экземпляра | Документация по вычислительному движку

В этой кодовой лаборатории мы настраиваем службу ретрансляции почты через Mailgun, выполнив следующие шаги:

1. Следуйте инструкциям в разделе «Отправка электронной почты с помощью Mailgun | Документация Compute Engine , чтобы настроить учетную запись в Mailgun и настроить службу ретрансляции электронной почты. Дополнительные модификации ниже.
2. Добавьте адреса электронной почты всех получателей в авторизованный список Mailgun. Этот список можно найти в разделе Mailgun>Отправка>Обзор на левой панели.

Как только получатели нажимают «Я согласен» в электронном письме, отправленном с адреса [e-mail] , их адреса электронной почты сохраняются в авторизованном списке для получения электронных писем с оповещениями по конвейеру.

3. Шаг 3 из раздела «Перед началом работы» — создайте правило брандмауэра следующим образом:

4. Шаг 3 «Настройка Mailgun в качестве ретранслятора почты с помощью Postfix». Выберите «Интернет-сайт» или «Интернет со смартхостом» вместо «Только локально» , как указано в инструкциях.

5. Шаг 4 «Настройка Mailgun в качестве ретранслятора почты с помощью Postfix». Отредактируйте vi /etc/postfix/main.cf, добавив 10.128.0.0/9 в конец mynetworks .

6. Отредактируйте vi /etc/postfix/master.cf, чтобы изменить smtp по умолчанию (25) на порт 587.

7. В правом верхнем углу студии Data Fusion нажмите «Настроить» . Нажмите «Предупреждение конвейера» и нажмите кнопку «+» , чтобы открыть окно «Предупреждения» . Выберите Отправить электронную почту .

8. Заполните форму настройки электронной почты . Выберите завершение, успех или неудачу в раскрывающемся списке «Условие выполнения» для каждого типа оповещения. Если Include Workflow Token = false , отправляется только информация из поля «Сообщение». Если включить токен рабочего процесса = true , информация из поля «Сообщение» и подробная информация о токене рабочего процесса не будут отправлены. Вы должны использовать строчные буквы для протокола . В качестве отправителя используйте любой « поддельный » адрес электронной почты, отличный от адреса электронной почты вашей компании.

7. Настройка, развертывание, запуск/планирование конвейера

1. В правом верхнем углу студии Data Fusion нажмите «Настроить» . Выберите Spark для конфигурации двигателя. Нажмите «Сохранить» в окне «Настроить».

2. Нажмите «Просмотр для просмотра данных»**,** и снова нажмите «Просмотр**», чтобы вернуться к предыдущему окну. Вы также можете **Запустить** конвейер в режиме предварительного просмотра.

3. Нажмите Журналы , чтобы просмотреть журналы.
4. Нажмите «Сохранить» , чтобы сохранить все изменения.
5. Нажмите «Импортировать» , чтобы импортировать сохраненную конфигурацию конвейера при создании нового конвейера.
6. Нажмите «Экспорт» , чтобы экспортировать конфигурацию конвейера.
7. Нажмите «Развернуть» , чтобы развернуть конвейер.
8. После развертывания нажмите «Выполнить» и дождитесь завершения работы конвейера.

Примечание. При запуске конвейера Cloud Data Fusion подготавливает временный кластер Cloud Dataproc, запускает конвейер, а затем удаляет кластер. Это может занять несколько минут. В течение этого времени вы можете наблюдать за состоянием перехода конвейера от подготовки к запуску , выполнению , деинициализации и успешному выполнению .

9. Вы можете дублировать конвейер, выбрав «Дублировать» под кнопкой «Действия» .
10. Вы можете экспортировать конфигурацию конвейера, выбрав «Экспорт» под кнопкой «Действия» .
11. Нажмите «Входящие триггеры» или «Исходящие триггеры» на левом или правом краю окна Studio, чтобы при необходимости настроить триггеры конвейера.
12. Нажмите «Расписание» , чтобы запланировать периодический запуск конвейера и загрузку данных.

13. Сводка показывает диаграммы истории запусков, записей, журналов ошибок и предупреждений.

8. Проверка

1. Конвейер проверки выполнен успешно.

2. Проверьте, есть ли в наборе данных BigQuery все таблицы.

bq ls $PROJECT_ID:$DATASET_ID

     tableId       Type    Labels   Time Partitioning
----------------- ------- -------- -------------------
 Allergies         TABLE
 Careplans         TABLE
 Conditions        TABLE
 Encounters        TABLE
 Imaging_Studies   TABLE
 Immunizations     TABLE
 Medications       TABLE
 Observations      TABLE
 Organizations     TABLE
 Patients          TABLE
 Procedures        TABLE
 Providers         TABLE

3. Получать оповещения по электронной почте (если настроено).

Просмотр результатов

Чтобы просмотреть результаты после запуска конвейера:

1. Запросите таблицу в пользовательском интерфейсе BigQuery. ПЕРЕХОДИТЬ В пользовательский интерфейс BIGQUERY
2. Обновите запрос ниже, указав собственное имя проекта, набор данных и таблицу.

9. Уборка

Чтобы избежать списания средств с вашей учетной записи Google Cloud Platform за ресурсы, используемые в этом руководстве:

После завершения обучения вы можете очистить ресурсы, созданные вами в GCP, чтобы они не занимали вашу квоту и вам не выставлялись счета за них в будущем. В следующих разделах описано, как удалить или отключить эти ресурсы.

Удаление набора данных BigQuery

Следуйте этим инструкциям, чтобы удалить набор данных BigQuery, созданный вами в рамках этого руководства.

Удаление сегмента GCS

Следуйте этим инструкциям, чтобы удалить сегмент GCS, созданный вами в рамках этого руководства.

Удаление экземпляра Cloud Data Fusion

Следуйте этим инструкциям, чтобы удалить экземпляр Cloud Data Fusion .

Удаление проекта

Самый простой способ избавиться от выставления счетов — удалить проект, созданный вами для этого руководства.

Чтобы удалить проект:

Внимание : Удаление проекта имеет следующие последствия:

• Все в проекте удалено. Если для этого руководства вы использовали существующий проект, то при его удалении вы также удаляете всю другую работу, выполненную в проекте.
• Пользовательские идентификаторы проектов теряются. Создавая этот проект, вы, возможно, создали собственный идентификатор проекта, который хотите использовать в будущем. Чтобы сохранить URL-адреса, использующие идентификатор проекта, например URL-адрес appspot.com , удалите выбранные ресурсы внутри проекта, а не удаляйте весь проект.

Если вы планируете изучить несколько руководств и кратких руководств, повторное использование проектов поможет вам избежать превышения ограничений квоты проекта.

1. В консоли GCP перейдите на страницу «Проекты» . ПЕРЕЙТИ НА СТРАНИЦУ ПРОЕКТОВ
2. В списке проектов выберите проект, который хотите удалить, и нажмите «Удалить» .
3. В диалоговом окне введите идентификатор проекта, а затем нажмите «Завершить работу» , чтобы удалить проект.

10. Поздравления

Поздравляем, вы успешно завершили лабораторную работу по кодированию для приема медицинских данных в BigQuery с помощью Cloud Data Fusion.

Вы импортировали данные CSV из Google Cloud Storage в BigQuery.

Вы визуально построили конвейер интеграции данных для массовой загрузки, преобразования и маскировки медицинских данных.

Теперь вы знаете ключевые шаги, необходимые для начала работы в области анализа данных здравоохранения с помощью BigQuery на Google Cloud Platform.