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

1. Введение

Последнее обновление: 28.02.2020

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

В этой практической работе по программированию вы узнаете:

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

Что вам понадобится для проведения этого практического занятия?

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

Если у вас нет проекта GCP, выполните следующие шаги для создания нового проекта GCP.

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

Если вам нужен новый набор данных, вы всегда можете сгенерировать его с помощью Synthea™ . Затем загрузите его в 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. Отправьте письмо на адрес [e-mail] с того адреса электронной почты, который вы используете для входа в Cloud Console, с просьбой о присоединении.
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

Помните: для этой лабораторной работы вы всегда можете сгенерировать новый набор реалистичных тестовых данных с помощью Synthea™ . Для вашего удобства мы предоставляем набор данных в GCS: gs://hcls_testing_data_fhir_10_patients/csv

3. Настройка среды облачного слияния данных

Выполните следующие шаги, чтобы включить 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, который появится в пользовательском интерфейсе Data Pipelines.

4. Наведите указатель мыши на узел Wrangler и нажмите «Свойства» . Нажмите кнопку Wrangler , затем выберите исходный файл .csv (например, patients.csv), который должен содержать все поля данных для построения нужной схемы.
5. Нажмите стрелку вниз (Преобразования столбцов) рядом с названием каждого столбца (например, body).
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 , который появится в пользовательском интерфейсе Data Pipelines.
3. Укажите на узел источника GCS и нажмите «Свойства» .

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

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

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

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

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

• Метка = {любой текст}
• Название поля ввода = {*}
• Precondition = {filename != "patients.csv"} для различения каждого входного файла ( например, patients.csv, providers.csv, allergys.csv и т. д. ) от узла Source.

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. Заполните обязательные поля. Установите следующие параметры:

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

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

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

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

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

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

Примечание: Для каждого входного CSV-файла создавайте отдельную ветку от исходного узла в конвейере.

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

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

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

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

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

Отправка электронных писем из экземпляра | Документация Compute Engine

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

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

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

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

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

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

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

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

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

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

1. В правом верхнем углу Data Fusion Studio нажмите «Настроить» . Выберите 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 в сфере анализа данных в здравоохранении.