Загрузка данных CSV (значения, разделенные запятыми) в BigQuery с помощью Cloud Data Fusion — прием в режиме реального времени

1. Введение

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

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

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

• Как загрузить CSV-данные (в режиме реального времени) из системы Pub/Sub в 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.

Публикация сообщений в Cloud Pub/Sub: https://cloud.google.com/pubsub/docs/publisher .

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™ . Для вашего удобства мы предоставляем набор данных в следующем хранилище Google Cloud Storage: gs://hcls_testing_data_fhir_10_patients/csv .

Установите и инициализируйте Google Cloud SDK , создайте публикации (Pub или Sub Topic) и подписки (Subscriptions).

gcloud init
gcloud pubsub topics create your-topic
gcloud pubsub subscriptions create --topic your-topic your-sub

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

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

Включить API .

1. Перейдите в библиотеку API консоли GCP .
2. Выберите свой проект из списка проектов.
3. В библиотеке API выберите API, который хотите включить ( Cloud Data Fusion API, Cloud Pub/Sub 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. Нажмите кнопку «+ Добавить еще одну роль» (или «Редактировать агент службы API Cloud Data Fusion»), чтобы добавить роль подписчика Pub/Sub.

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 Data Fusion в GCP, давайте начнем создавать конвейеры обработки данных в Cloud Data Fusion, выполнив следующие шаги:

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

3. Нажмите зеленую кнопку «+» в правом верхнем углу, затем выберите «Создать конвейер». Или нажмите ссылку «Создать» для создания конвейера.

4. После появления окна «Студия конвейера» в левом верхнем углу выберите в выпадающем списке «Конвейер данных — в реальном времени» .

5. В пользовательском интерфейсе конвейеров данных на левой панели вы увидите различные разделы: Фильтр, Источник, Преобразование, Аналитика, Приемник, Обработчики ошибок и Оповещения, где вы можете выбрать узел или узлы для конвейера.

Выберите узел «Источник» .

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

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

• Метка = {любой текст}
• Название ссылки = {любой текст}
• Идентификатор проекта = автоматическое определение
• Подписка = Подписка, созданная в разделе «Создать тему публикации/подписки» (например, your-sub ).
• Тема = Тема, созданная в разделе «Создать тему публикации/подтемы» (например, ваша тема ).

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

5. Чтобы закрыть окно «Свойства публикации/подписки», нажмите кнопку «X» .

Выберите узел «Преобразование» .

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

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

• Convert = преобразовать сообщение из байтового типа в строковый.
• Поля для удаления = {любое поле}
• Поля для сохранения : { сообщение , метка времени и атрибуты } ( например, атрибуты: ключ='имя файла':значение='пациенты', отправленные из Pub/Sub)
• Поля для переименования = { сообщение , метка времени }

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

5. В разделе «Преобразование» на панели плагинов слева дважды щелкните узел Wrangler , который появится в пользовательском интерфейсе Data Pipelines. Соедините узел преобразования Projection с узлом преобразования Wrangler. Наведите указатель мыши на узел Wrangler и щелкните «Свойства» .

6. Нажмите на выпадающее меню «Действия» и выберите «Импорт» , чтобы импортировать сохраненную схему (например: gs://hcls_testing_data_fhir_10_patients/csv_schemas/ schema (Patients).json ).
7. Добавьте поле TIMESTAMP в схему вывода (если оно отсутствует), нажав кнопку «+» рядом с последним полем и установив флажок «Null».
8. Заполните обязательные поля. Установите следующие параметры:

• Метка = {любой текст}
• Название поля ввода = {*}
• Предварительное условие = { attributes.get("filename") != "patients" } для различения каждого типа записей или сообщений ( например, пациенты, поставщики услуг, аллергии и т. д. ), отправляемых из исходного узла PubSub.

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

10. Укажите желаемый порядок названий столбцов и удалите ненужные поля. Скопируйте следующий фрагмент кода и вставьте его в поле «Рецепт».

drop attributes
parse-as-csv :body ',' false
drop body
set columns TIMESTAMP,Id,BIRTHDATE,DEATHDATE,SSN,DRIVERS,PASSPORT,PREFIX,FIRST,LAST,SUFFIX,MAIDEN,MARITAL,RACE,ETHNICITY,GENDER,BIRTHPLACE,ADDRESS,CITY,STATE,ZIP
mask-number SSN xxxxxxx####

Примечание: Значение «False» означает, что анализ столбца с содержимым выполняется без отмеченной опции «Установить первую строку в качестве заголовка».

11. Для маскирования и обезличивания данных обратитесь к руководству Batch-Codelab - CSV to BigQuery via CDF . Или добавьте этот фрагмент кода mask-number SSN xxxxxxx#### в поле «Рецепт».
12. Чтобы закрыть окно «Свойства преобразования», нажмите кнопку «X» .

Выберите узел «Приёмник».

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

3. Заполните обязательные поля:

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

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

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

5. Создайте конвейер обработки данных в реальном времени.

В предыдущем разделе мы создали узлы, необходимые для построения конвейера обработки данных в Cloud Data Fusion. В этом разделе мы соединим узлы для построения самого конвейера.

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

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

Примечание: Для каждого сообщения PubSub в конвейере создавайте отдельную ветвь от узла Projection.

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

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

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

Отправляйте сообщения через Cloud Pub/Sub.

Использование пользовательского интерфейса публикации/подписки :

1. Перейдите в консоль GCP -> Публикация/Подписка -> Темы, выберите свою тему , затем нажмите ОПУБЛИКОВАТЬ СООБЩЕНИЕ в верхнем меню.

2. В поле «Сообщение» указывайте только одну строку записи за раз. Нажмите кнопку «+ДОБАВИТЬ АТРИБУТ» . Укажите ключ = имя файла , значение = < тип записи > ( например, пациенты, поставщики услуг, аллергии и т. д. ).
3. Нажмите кнопку «Опубликовать», чтобы отправить сообщение.

Используя команду gcloud :

1. Введите сообщение вручную.

gcloud pubsub topics publish <your-topic> --attribute <key>=<value> --message \
"paste one record row here"

2. Полуавтоматически передавайте сообщение с помощью команд Unix cat и sed . Эту команду можно запускать многократно с различными параметрами.

gcloud pubsub topics publish <your-topic> --attribute <key>=<value> --message \
"$(gsutil cat gs://$BUCKET_NAME/csv/<value>.csv | sed -n '#p')"

Примечание: <ваша-тема> - та же тема, что указана в свойствах исходного узла Pub/Sub.

<ключ> - имя файла

cat - чтение содержимого файла

sed - подавляет вывод ( -n ) и явно выводит ( p ) конкретную строку #(1-?). Если строка заголовка существует, используйте '2p' и так далее.

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

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

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

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

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

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

12. Нажмите «Сводка» , чтобы отобразить диаграммы истории выполнения, записи, журналы ошибок и предупреждения.

7. Валидация

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

1. Убедитесь, что конвейер был успешно выполнен и работает непрерывно.

2. Убедитесь, что таблицы BigQuery загружены обновленными записями на основе временной метки (TIMESTAMP). В этом примере две записи или сообщения о пациентах и ​​одна запись или сообщение об аллергии были опубликованы в тему Pub/Sub 25 июня 2019 года.

bq query --nouse_legacy_sql 'select (select count(*) from \
'$PROJECT_ID.$DATASET_ID.Patients'  where TIMESTAMP > "2019-06-25 \
01:29:00.0000 UTC" ) as Patients, (select count(*) from \
'$PROJECT_ID.$DATASET_ID.Allergies' where TIMESTAMP > "2019-06-25 \
01:29:00.0000 UTC") as Allergies;'

Waiting on bqjob_r14c8b94c1c0fe06a_0000016b960df4e1_1 ... (0s) Current status: DONE  
+----------+-----------+
| Patients | Allergies |
+----------+-----------+
|        2 |         1 |
+----------+-----------+

3. Убедитесь, что сообщения, опубликованные в теме <your-topic>, были получены подписчиком <your-sub>.

gcloud pubsub subscriptions pull --auto-ack <your-sub>

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

Чтобы просмотреть результаты после публикации сообщений в тему Pub/Sub во время работы конвейера обработки в реальном времени:

1. Выполните запрос к таблице в пользовательском интерфейсе BigQuery. ПЕРЕЙДИТЕ В ИНТЕРФЕЙС BIGQUERY
2. Обновите приведенный ниже запрос, указав название вашего проекта, набор данных и таблицу.

8. Уборка

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

9. Поздравляем!

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

Вы опубликовали данные в формате CSV в топик Pub/Sub, а затем загрузили их в BigQuery.

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

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