Лучший опыт

Инженерия данных — не только для инженеров!.

“Ненавижу это!”  —  именно таким было мое отношение к первому заданию по инженерии данных. Мне поручили выполнить веб-скрейпинг нескольких цифровых данных из Википедии. Пришлось уговаривать себя: “Сделай эту работу на Python! Ведь времена ручного поиска данных позади. Запасись терпением. У тебя оно есть”. Проблема: может ли выпускник факультета общественных наук профессионально заниматься инженерией данных? К примеру, создать авт
Инженерия данных — не только для инженеров!...

“Ненавижу это!”  —  именно таким было мое отношение к первому заданию по инженерии данных. Мне поручили выполнить веб-скрейпинг нескольких цифровых данных из Википедии.

Пришлось уговаривать себя: “Сделай эту работу на Python! Ведь времена ручного поиска данных позади. Запасись терпением. У тебя оно есть”.

Проблема: может ли выпускник факультета общественных наук профессионально заниматься инженерией данных? К примеру, создать автоматизированный конвейер данных и надежно защитить его в облаке?

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

Стоит ли отклонить эту гипотезу? Мое шестое чувство подсказывает: “Да”. Быть не может, чтобы такой непрофессионал в области ИТ, как я, смог этого добиться.

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

Бизнес-кейс

Моя задача заключалась в создании базы данных для компании Gans, которая предоставляет для транспортировки электромобили. Чтобы оптимизировать предложение достаточного количества средств передвижения в определенное время, компании нужен специалист по изучению данных, который создаст базу данных для измерения спроса в зависимости от погодных факторов.

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

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

В будущем базу данных можно усовершенствовать, включив в нее данные о прибытии общественного транспорта (например, с помощью API для авиарейсов, поездов и автобусов). Это позволит получить более полное представление о прибытии туристов (некоторые из них станут потенциальными клиентами компании).

Вот как выглядит карта проекта, включая API погоды (Weather) и авиарейсов (Flights):

В данной статье рассмотрим только интеграцию погодных данных.

Начнем с импортирования всех библиотек, необходимых для работы в Python:

import pandas as pd
import requests   
from bs4 import BeautifulSoup 
from datetime import datetime     
from pytz import timezone

Веб-скрейпинг 101: как получить HTML-код?

Первый шаг: создание таблицы с данными о городах. Эта таблица будет служить основным связующим звеном с остальными таблицами. Она будет статичной, поскольку содержащиеся в ней данные не придется регулярно обновлять. Назовем эту таблицу “cities_info” (“информация о городах”).

Для каждого города в ней будет одна строка и пять столбцов:

  • уникальный идентификатор города (city ID);
  • название города (city’s name);
  • земля Германии, к которой относится город (German state);
  • широта (latitude);
  • долгота (longitude).

Эти данные по каждому городу можно получить с помощью веб-скрейпинга с сайтов английской Википедии. Достаточный бюджет позволил бы купить API городов с достоверными данными и методологической информацией. Но в рамках этого проекта буду отрабатывать навыки скрейпинга, используя Википедию.

Меня пугала мысль о том, что для каждого сайта придется вручную экспортировать HTML-код и копировать его в Vs Code. К счастью, с помощью библиотеки “requests” на Python это можно делать автоматически:

import requests
berlin_url = "https://en.wikipedia.org/wiki/Berlin"
berlin_response = requests.get(berlin_url) 
berlin_soup = BeautifulSoup(berlin_response.content, 'html.parser') 
print(berlin_soup.prettify)

1:0 в пользу Python.

Циклы и функции

Циклы и функции  —  неотъемлемая часть итерации. Без них невозможно воспроизвести один и тот же код и одновременно применить его к разным URL. Это ключ к автоматизации веб-скрейпинга.

Вот пример цикла, использованный мной для получения HTML-кода каждого сайта Википедии:

cities_list = ["Berlin", "Hamburg", "Munich", "Cologne", "Frankfurt"]

for city in cities_list:
  url = f"https://www.wikipedia.org/wiki/{city}"   # превращает url в строку f и помещает город в качестве переменной, которая меняется в зависимости от названия города
  response = requests.get(url)          # получает все содержимое страницы Википедии и сохраняет его под именем response
  city_soup = BeautifulSoup(response.content, 'html.parser')

Пока еще функции меня пугают, поэтому стараюсь избегать их по мере возможности.

1000:0 в пользу Python.

Как получить доступ к материалам в формате HTML?

В HTML есть так называемые “p-теги”. Получить к ним доступ позволит следующий код:

print(soup.p)   # получает доступ к первому p-тегу 
print(soup.p.string)  # получает доступ к строке, связанной с первым p-тегом 
for child in soup.div:  # пример поиска и вывода каждого дочернего элемента в 1-м div
    print(child)

Создание датафреймов в Python и соответствующих таблиц в SQL

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

Первый датафрейм работает как статический индекс (то есть его информация не будет регулярно обновляться): cities_info.

cities_list = ["Berlin", "Hamburg", "Munich", "Cologne", "Frankfurt"]
states = []
latitudes = []
longitudes = []

for city in cities_list:
  url = f"https://www.wikipedia.org/wiki/{city}"   
  city_soup = BeautifulSoup(response.content, 'html.parser')    # парсит контент (сохраняет содержимое сайта Википедии в переменной city_soup)

  # получение земли, к которой принадлежит город
  if city not in ["Hamburg", "Berlin"]:         # для Берлина тоже подойдет общая формула .find! Только для Гамбурга нет раздела "State".
    city_state = city_soup.find("table", class_="vcard").find(string="State").find_next("td").get_text()  # получает данные о земле для городов с другим названием земли
  else:
    city_state = city     # в случае Гамбурга и Берлина просто берется одноименное название города
  states.append(city_state)

  # получение широты каждого города и добавление ее в столбец latitudes
  city_latitude = city_soup.find(class_="latitude").get_text()
  latitudes.append(city_latitude)  

   # получение долготы каждого города и добаление ее в столбец longitude
  city_longitude = city_soup.find(class_="longitude").get_text()
  longitudes.append(city_longitude)  


cities_info_non_rel = pd.DataFrame({         # это будет датафрейм cities_info
    "city_name": cities_list,
    "german_state": states,
    "latitude": latitudes,
    "longitude": longitudes
})  

display(cities_info_non_rel)    # вместо вывода показывается таблица в более красивом формате

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

populations = []
year_data_retrieved = []

for city in cities_list:
  url = f"https://www.wikipedia.org/wiki/{city}"   # превращает url в строку f и помещает город в качестве переменной, так что она меняется в соответствии с названием города 
  response = requests.get(url)          # получает все содержимое страницы Википедии и сохраняет его под именем response
  city_soup = BeautifulSoup(response.content, 'html.parser')    # парсит контент  (сохраняет содержимое сайта Википедии в переменной city_soup)

# получение количества жителей для каждого города и добавление его в столбец population
  city_population = int(city_soup.find("table", class_="vcard").find(string="Population").find_next("td").get_text().replace(",", ""))  # находит численность населения, очищает и превращает в целое число
  populations.append(city_population)  

  # запись года, когда были получены данные
  current_year = (datetime.now()).year
  year_data_retrieved.append(current_year)

cities_population = pd.DataFrame({         # это будет датафрейм cities_population
    "city_name": cities_list,
    "population": populations,
    "year_data_retrieved": year_data_retrieved,
    })  

display(cities_population)

Теперь в MySQL можно создать две таблицы, которые будут содержать всю информацию, впоследствии передаваемую из Python и облака. Это значит, что нужны таблицы с таким же количеством столбцов и типов данных, как в созданных датафреймах Python:

/***************************
Настройка среды
***************************/

-- Удалите базу данных, если она уже существует
DROP DATABASE IF EXISTS gans;

-- Создайте базу данных
CREATE DATABASE gans;

-- Используйте базу данных
USE gans;


/***************************
Создание первой таблицы
***************************/

-- Создайте таблицу 'cities_info'
CREATE TABLE cities_info (
    cities_id INT AUTO_INCREMENT, -- Автоматически генерируемый идентификатор для каждого города
    city_name VARCHAR(255) NOT NULL, -- Название города
    german_state VARCHAR(255) NOT NULL, -- Название земли 
    latitude VARCHAR(255) NOT NULL, -- Координата широты
    longitude VARCHAR(255) NOT NULL, -- Коордианта долготы 
    PRIMARY KEY (cities_id) -- Первичный ключ для уникальной идентификации каждого города
    );


/***************************
Создание второй таблицы
***************************/ 

CREATE TABLE cities_population (
 cities_id INT,
 population INT, -- Численность населения
    year_data_retrieved INT, -- год получения данных о населении
    FOREIGN KEY (cities_id) REFERENCES cities_info(cities_id)
    );

Переносим данные из Python в первую таблицу SQL:

schema = "gans"     # название базы данных
host = "xxx.x.x.x"
user = "root"               # имя пользователя
password = "xxxx"           # явно укажите пароль SQL или импортируйте из другого ноутбука ("from xxxfile import my_password")
port = 3306

connection_string = f'mysql+pymysql://{user}:{password}@{host}:{port}/{schema}'    
# Это то, что соединяет ноутбук python с sql workbench
cities_info_non_rel.to_sql('cities_info',   # Таким образом мы передаем данные от PYTHON к SQL
                  if_exists='append',       # не нужно ничего перезаписывать, просто добавляем данные к уже существующим
                  con=connection_string,    # con - это аргумент, который нужен для подключения к sql workbench
                  index=False)

Обратите внимание: первая попытка выполняется локально. Позже, добавив экземпляр Google Cloud Platform, можно будет отредактировать поле “host” и напрямую отправить эти данные в облако.

Создав первую таблицу в SQL, получаем данные, содержащиеся в cities_info, чтобы использовать столбец cities_id в качестве индекса во втором датафрейме cities_population:

cities_info = pd.read_sql("cities_info", con=connection_string)   # "считывает" информацию, хранящуюся в SQL
cities_info

Вот как выглядит cities_info:

Прежде чем передать содержимое второго датафрейма в SQL, используем вновь созданный столбец cities_id, добавив его в датафрейм cities_populations.

Изменяем порядок столбцов и удаляем столбец city_name, который больше не понадобится, поскольку теперь в датафрейме cities_populations добавлен столбец cities_id:

# добавляем столбец cities_id в другой датафрейм population
cities_population["cities_id"] = cities_info_non_rel["cities_id"]
cities_population

#  убираем столбец city_name, потому что он нам больше не нужен + меняем порядок столбцов, чтобы сделать процесс более интуитивным
cities_population = cities_population[["cities_id", "population", "year_data_retrieved"]]
cities_population

Вот и второй датафрейм, который будет частично динамическим (его данные будут обновляться время от времени, примерно раз в год):

На данный момент будем считать cities_population нединамической таблицей, поскольку она не будет обновляться до следующего года.

Теперь можно передать ее содержимое в локальный экземпляр SQL:

# передааем содержимое второй таблицы в SQL
cities_population.to_sql('cities_population',   # Таким образом мы передаем данные от PYTHON к SQL
                  if_exists='append',       # не нужно ничего перезаписывать, просто добавляем данные к уже существующим
                  con=connection_string,    # con - это аргумент, который нужен для подключения к sql workbench
                  index=False)

После запуска всего этого в SQL и использования функции “Reverse engineer” (“реверс-инжиниринг”), получим начальную версию схемы:

Схема 101

Погружаемся в процесс обучения еще глубже.

Следующий шаг: интеграция API в базу данных. Это простейшая операция, которую может выполнить любой дата-сайентист.

Лучше освоить ее, пока ИИ не научится делать это за нас.

Использование API прогнозов погоды

Третий важный шаг этого проекта  —  извлечение погодных данных из существующего хранилища данных и интеграция их в базу данных SQL. Доступ к таким хранилищам осуществляется через API (Application Programming Interface  —  интерфейс прикладного программирования). API  —  это набор правил и протоколов, которые позволяют различным программным приложениям взаимодействовать друг с другом.

Воспользуемся сайтом openweathermap.org, который позволит получить доступ к бесплатным прогнозам погоды для любой точки мира. Будем использовать API прогнозов погоды на 5 дней, включая данные прогноза с 3-часовым интервалом.

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

def get_weather_data(cities):
  # Поскольку мы будем работать в облаке, компьютеры могут находиться где угодно. Исправим часовой пояс компьютера на наш местный часовой пояс
  berlin_timezone = timezone("Europe/Berlin")
  API_key = "7e5623c79f102b6c08b15c8hjib4cc9l"    # это не настоящий ключ
  weather_items = []

  for city in cities:
    url = (f"http://api.openweathermap.org/data/2.5/forecast?q={city}&appid={API_key}&units=metric")
    response = requests.get(url)
    json = response.json()

    # Добавлено время получения, чтобы мы знали, когда был сделан прогноз
    retrieval_time = datetime.now(berlin_timezone).strftime("%Y-%m-%d %H:%M:%S")

    # Поскольку сейчас мы используем данные из реляционной базы данных, город должен обозначать city_id, а не название города
    city_id = cities_info.loc[cities_info["city_name"] == city, "city_id"].values[0]  # здесь нам нужно получить значения, иначе будут показаны серии 

    for item in json["list"]:
        weather_item = {
            # Добавлено название города, чтобы информация была понятна при просмотре нескольких городов
            "forecast_time": item.get("dt_txt", None),
            "temperature": item["main"].get("temp", None),
            "feels_like": item["main"].get("feels_like", None),
            "forecast": item["weather"][0].get("main", None),
            "humidity": item["main"].get("humidity", None),
            "rain_in_last_3h": item.get("rain", {}).get("3h", 0),
            "risk_of_rain": item["pop"],
            "snow_in_last_3h": item.get("snow", {}).get("3h", 0),      
            "wind_speed": item["wind"].get("speed", None),
            "data_retrieved_at": retrieval_time
        }

        weather_items.append(weather_item)
  
  weather_df = pd.DataFrame(weather_items)
  weather_df["forecast_time"] = pd.to_datetime(weather_df["forecast_time"])
  weather_df["data_retrieved_at"] = pd.to_datetime(weather_df["data_retrieved_at"])
  weather_df["snow_in_last_3h"] = pd.to_numeric(weather_df["snow_in_last_3h"], downcast="float")

  return weather_df

weather_df = get_weather_data(["Berlin", "Hamburg", "Munich", "Cologne", "Frankfurt"])
weather_df     # создадим новый датафрейм с помощью функции

Одна из ключевых особенностей таблицы и датафрейма с информацией о погоде заключается в том, что они должны содержать столбец с указанием времени получения данных. Это позволит, помимо прочего, осуществлять соответствующую фильтрацию и создавать процедуры или функции на SQL для автоматического удаления устаревших данных спустя какое-то время. Это также необходимо для обеспечения согласованности данных, чтобы различать прогнозируемые дату/время (forecast time) от даты/времени получения данных (data retrieval).

Вот как выглядит новый погодный датафрейм на Python:

Получив структуру погодного датафрейма, перейдем к созданию третьей таблицы с именем weather (погода) на SQL:

/***************************
Создание таблицы WEATHER
***************************/ 

CREATE TABLE weather (
    city_id INT,
    forecast_time datetime,
    temperature float,
    feels_like float,
    forecast VARCHAR(255) NOT NULL,
    humidity INT,
    rain_in_last_3h FLOAT,
    risk_of_rain FLOAT,
    snow_in_last_3h FLOAT,
    wind_speed FLOAT,
    data_retrieved_at DATETIME,
    FOREIGN KEY (city_id) REFERENCES cities_info(city_id)
    );

-- TRUNCATE TABLE weather; -- нам может понадобиться стереть все строки, когда таблица станет слишком длинной, а данные слишком старыми

Это обновленная схема SQL после реверс-инжиниринга:

Переход от локального к глобальному

Прежде чем переносить код в облако, нужно убедиться, что код получения погодных данных работает локально. Для этого сразу после создания кода на Python попробуем перенести его в локальный экземпляр SQL:

weather_df.to_sql("weather",           
                  if_exists='append',       
                  con=connection_string,
                  index=False)

Если в таблицу SQL будут добавляться новые данные, значит, она работает!

Вот как выглядела таблица погоды SQL после локальной подачи данных из Python:

Интеграция с Google Cloud Platform

Убедившись, что код работает локально, перенесем его в облако.

Откроем аккаунт Google Cloud Platform (GCP) и настроим облачный экземпляр. Пошаговое описание этого процесса на примере MySQL можно найти здесь.

Одним из самых важных шагов является изменение локального IP-хоста на IP-хост Google в ноутбуке Python, в частности, в блоке кода, который передает данные в SQL:

schema = "gans"         # название базы данных 
host = "XX.XXX.XX.XX"   # LOCAL HOST:"xxx.x.x.x" (до того, как я изменил это на облако)
user = "root"           # имя пользователя (см. инструкции)
password = "XXXX"       # явно укажите пароль или импортируйте из другого ноутбука ("from xxxfile import my_password")
port = 3306

connection_string = f'mysql+pymysql://{user}:{password}@{host}:{port}/{schema}'     # Это то, что соединяет ноутбук  python с sql workbench

Таким образом, ранее созданные статические таблицы на Python могут быть автоматически загружены в облако и заполнить уже существующие таблицы SQL без необходимости выполнять код на Python когда-либо снова. По крайней мере, пока не понадобится обновить нединамические таблицы (cities_info и cities_population).

Для целей данного проекта достаточно загрузить эти две таблицы из Python в GCP, изменив IP-адрес хоста. Другой вариант  —  загрузить код, создающий эти две таблицы, в облако. В данном случае поместим в облако только код, который создает и заполняет динамическую таблицу погоды.

После создания функции в поле “Cloud Functions” (“Облачные функции”) в GCP методом множества проб и ошибок код наконец-то заработал:

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

  1. Перейдите на Cloud Run, кликнув на название функции в этом боксе.
  2. Кликните на “Edit and deploy new version”.
  3. Прокрутите до низа контейнера (ов).
  4. Находясь в Cloud SQL Connections, нажмите Add Connection.
  5. Включите API админа в Cloud SQL.
  6. Выберите свою базу данных из выпадающего списка (если появится бокс с просьбой включить API, включите этот API).
  7. Нажмите “Deploy”.
  8. Когда функция будет развернута, нажмите снова на URL. Вы должны успешно отправить данные в базу данных SQL. Если все прошло удачно, вы увидите сообщение “Data successfully added”.

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

Данные успешно добавлены

По сравнению с кодом, протестированным локально, мне потребовалось внести несколько изменений. Вот как выглядит окончательный вариант кода:

import functions_framework
import pandas as pd
import sqlalchemy
import requests  
from pytz import timezone
from datetime import datetime 

@functions_framework.http
def insert(request):
  connection_string = connection()
  insert_into_weather(connection_string)
  return 'Data successfully added'

def connection():
  connection_name = "flying-dove-416317:europe-west1:wbs-mysql-db"    # это не настоящие данные 
  db_user = "root"
  db_password = "xxxx"   # введите свой пароль SQL
  schema_name = "gans"

  driver_name = 'mysql+pymysql'
  query_string = {"unix_socket": f"/cloudsql/{connection_name}"}

  db = sqlalchemy.create_engine(
      sqlalchemy.engine.url.URL(
          drivername = driver_name,
          username = db_user,
          password = db_password,
          database = schema_name,
          query = query_string,
      )
  )
  return db


# ЗДЕСЬ ЗАПУСКАЕТСЯ ФУНКЦИЯ ПОЛУЧЕНИЯ ПОГОДНЫХ ДАННЫХ:

def extract_city(connection_string):
    return pd.read_sql("cities_info", con=connection_string)

def get_weather_data(cities_df):
  berlin_timezone = timezone("Europe/Berlin")
  API_key = "7e5623c79f102b6c08b15c8hjib4cc9l"    # это не настоящий ключ
  weather_items = []
    
  for city in cities_df["city_name"]:
    url = (f"http://api.openweathermap.org/data/2.5/forecast?q={city}&appid={API_key}&units=metric")
    response = requests.get(url)
    json = response.json()

    # Добавлено время получения, чтобы мы знали, когда был сделан прогноз
    retrieval_time = datetime.now(berlin_timezone).strftime("%Y-%m-%d %H:%M:%S")

    # Поскольку сейчас мы используем данные из реляционной базы данных, город должен обозначать city_id, а не название города 
    city_id = cities_df.loc[cities_df["city_name"] == city, "city_id"].values[0]  # здесь нам нужно получить значения, иначе будут показаны серии

    for item in json["list"]:
        weather_item = {
            # Добавлено название города, чтобы информация была понятна при просмотре нескольких городов
            "forecast_time": item.get("dt_txt", None),
            "temperature": item["main"].get("temp", None),
            "feels_like": item["main"].get("feels_like", None),
            "forecast": item["weather"][0].get("main", None),
            "humidity": item["main"].get("humidity", None),
            "rain_in_last_3h": item.get("rain", {}).get("3h", 0),
            "risk_of_rain": item["pop"],
            "snow_in_last_3h": item.get("snow", {}).get("3h", 0),      
            "wind_speed": item["wind"].get("speed", None),
            "data_retrieved_at": retrieval_time
        }

        weather_items.append(weather_item)
  
  weather_df = pd.DataFrame(weather_items)
  weather_df["forecast_time"] = pd.to_datetime(weather_df["forecast_time"])
  weather_df["data_retrieved_at"] = pd.to_datetime(weather_df["data_retrieved_at"])
  weather_df["snow_in_last_3h"] = pd.to_numeric(weather_df["snow_in_last_3h"], downcast="float")

  return weather_df

def insert_into_weather(connection_string):
  cities_df = extract_city(connection_string)
  weather_df = get_weather_data(cities_df)    #создадим новый датафрейм с помощью функции
  weather_df.to_sql("weather", 
            if_exists="append",
            con=connection_string,
            index=False)

Уроки, усвоенные при создании кода, пригодного для работы в облаке

1. Зависимости

Одно из основных условий правильной работы кода  —  добавление необходимых зависимостей в файл requirements.txt. Важно учитывать, что некоторые библиотеки уже загружены на GCP по умолчанию и не должны включаться в .txt-файл, но их необходимо добавить в исходный код в качестве библиотек, например:

from pytz import timezone
from datetime import datetime

При попытке развертывания можно столкнуться с ошибкой, помеченной красным флагом (как показано на изображении ниже). В большинстве случаев это связано с импортом библиотек, не указанных в файле requirements.txt.

Текст ошибки: “Эту функцию не удалось развернуть, поэтому она не будет работать корректно. Отредактируйте ее и повторите развертывание”

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

Еще один полезный совет  —  выводите все внешние библиотеки, используемые в коде. После вызова функций в ноутбуке нужно выполнить:

print('\n'.join(f'{m.__name__}=={m.__version__}' for m in globals().values() if getattr(m, '__version__', None)))

Затем можно напрямую скопировать и вставить в файл requirements.txt нужные зависимости:

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

functions-framework==3.*
SQLAlchemy==1.4.37
PyMySQL==1.0.2
pandas==1.5.2
requests==2.31.0

2. Тестирование кода подключения

Перед добавлением функций получения данных следует сначала проверить, работает ли код подключения. Это позволит убедиться в том, что проблема кроется в коде, а не в настройке соединения.

Лучший способ начать перенос локального скрипта в GCP  —  создать “фиктивную” функцию и протестировать ее. Например:

import functions_framework
import pandas as pd
import sqlalchemy

@functions_framework.http
def insert(request):
  connection_string = connection()
  insert_into_test_table(connection_string)
  return 'Data successfully added'

def connection():
  connection_name = "YOUR_DB_CON_NAME"  # изменить соответственно
  db_user = "root"                      # изменить
  db_password = "YOUR_PASSWORD"         # изменить
  schema_name = "test_schema"           # изменить

  driver_name = 'mysql+pymysql'
  query_string = {"unix_socket": f"/cloudsql/{connection_name}"}

  db = sqlalchemy.create_engine(
      sqlalchemy.engine.url.URL(
          drivername = driver_name,
          username = db_user,
          password = db_password,
          database = schema_name,
          query = query_string,
      )
  )
  return db

def insert_into_test_table(con_str):
  data = {'FirstName': ['Function', 'Test'],
          'City': ['Cloud', 'Complete']}
  df = pd.DataFrame(data)
  df.to_sql(name="test_table", con=con_str, if_exists='append', index=False)

Если эта простая функция работает, значит, и соединение работает. Остается только добавить и интегрировать фактический код.

Если получена приведенная ниже ошибка, проблема наверняка в коде, а не в соединении:

500 Internal Server Error: сервер столкнулся с внутренней ошибкой и не смог выполнить ваш запрос. Либо сервер перегружен, либо в приложении произошла ошибка.

3. Проблемы с реальным кодом

Основная ошибка, допущенная мной при размещении кода в сети, заключалась в следующем: код был настроен на основе “фиктивного” шаблона, и функция (“insert_into_weather”), вызванная в разделе “@functions_framework.http”, не была определена.

После добавления этой функции код по-прежнему не работал. Это было связано с тем, что нужные города были названы мной вручную, а не взяты из датафрейма cities_info. Пришлось добавить в начало блока еще одну функцию под названием “extract_city” (“извлечение города”) и убедиться, что весь блок работает локально, прежде чем выкладывать его в сеть.

После использования таблицы cities_info для извлечения названий городов я допустил еще одну ошибку. Она заключалась в том, что не был обновлен следующий цикл, который должен был уже брать названия городов из датафрейма вместо прежнего списка и не мог найти названия с прежними аргументами.

Последние шаги

Чтобы сделать проект полностью функциональным и автоматизированным, нужно запланировать периодический запуск функции. Это довольно просто сделать с помощью меню “Cloud Scheduler” (“Облачный планировщик”) на GCP.

Для этого нужно нажать “Schedule a job” (“Запланировать задание”), указать несколько параметров, задав периодичность (раз в неделю или в определенное время в определенный день, используя Cron Expressions), чтобы функция запускалась в указанное время и заполняла SQL-таблицу необходимыми данными. У меня запуск был настроен на 15:00 каждого понедельника:

Убедимся, что это работает на SQL:

Прежде чем распрощаться с первым проектом SQL-Python-GCP, следует удалить свои данные из аккаунта GCP, чтобы созданный проект не расходовал ваши бесплатные кредиты. Речь идет об облачном экземпляре, функции и запланированном задании.

Заключение

У меня получилось создать базовую БД, используя SQL, Python и облачные вычисления.

В начале работы над проектом мне ничего не было известно ни о Google Cloud Platform, ни о любой другой платформе облачных вычислений. До сих пор я ничего не знаю о серверах, хостах, экземплярах и программных соединениях. Мои знания пока ограничиваются основами SQL и Python.

Как новичка, меня сначала одолевали сомнения: смогу ли справиться с такой сложной задачей, как создание автоматизированной базы данных и периодическое наполнение ее ценными данными? Теперь эти сомнения развеялись!

В дальнейшем постараюсь создать более полный и полезный конечный продукт, получая данные из различных API и объединяя их в своей функции GCP.

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

Генерация ответа, дополненная результатами поиска (RAG): от теории к реализации на LangChain.
Псевдоним типа в Swift.