Запуск функции Python в BigQuery

Ответ 1

Если вы хотите включить функцию Python в сам SQL, вот пример Эда Фримена, который представляет собой сценарий, который отображает день с 1 7 на день недели с Mon Sun.

Входные данные вставляются с помощью следующей команды:

DROP TABLE IF EXISTS DaysOfWeek

CREATE TABLE DaysOfWeek (
    [DayOfWeek] nvarchar(10) NOT NULL
)
GO

INSERT INTO DaysOfWeek VALUES
('Sunday'),
('Monday'),
('Tuesday'),
('Wednesday'),
('Thursday'),
('Friday'),
('Saturday'),
('Friday'),
('Monday'),
('Sunday')
GO

Запрос на отображение дня выглядит следующим образом:

EXEC sp_execute_external_script 
@language = N'Python',
@script = N'
OutputDataSet = InputDataSet
global daysMap
daysMap = {
    "Monday" : 1,
    "Tuesday" : 2,
    "Wednesday" : 3,
    "Thursday" : 4,
    "Friday" : 5,
    "Saturday" : 6,
    "Sunday" : 7
}
OutputDataSet["DayOfWeekNumber"] = pandas.Series([daysMap[I] for i in OutputDataSet["DayOfWeek"]], index = OutputDataSet.index, dtype = "int32")
    ', 
@input_data_1 = N'SELECT * FROM DaysOfWeek',
@input_data_1_name = N'InputDataSet'
WITH RESULT SETS (("DayOfWeek" nvarchar(10) null,"DayOfWeekNumber" int null))

Но больший скрипт будет выглядеть грязно внутри хранимой процедуры. Поэтому нам нужно упаковать файл Python и затем импортировать его в сценарий. Итак, создайте папку и вставьте пустой файл __init__.py вместе с вашими файлами Python, которые содержат определения. Эта папка должна быть расположена где-то, что SQL может увидеть. Затем с помощью команды sys.path.append добавьте пользовательский каталог в системные пути, чтобы интерпретатор мог найти файл. После помещения упакованных файлов в папку PYTHON_SERVICES внутри каталога экземпляра SQL все, что вам нужно сделать, это импортировать модуль и вызвать его определение.

Определение выглядит так:

def map_days(InputDataSet):

    import pandas
    OutputDataSet = InputDataSet

    global daysMap

    daysMap = {
        "Monday" : 1,
        "Tuesday" : 2,
        "Wednesday" : 3,
        "Thursday" : 4,
        "Friday" : 5,
        "Saturday" : 6,
        "Sunday" : 7
    }

OutputDataSet["DayOfWeekNumber"] = pandas.Series([daysMap[i] for i in OutputDataSet["DayOfWeek"]], index = OutputDataSet.index, dtype = "int32")
return OutputDataSet

Затем вызовите его прямо в SQL:

EXEC sp_execute_external_script 
@language = N'Python',
@script = N'
from MyPackages.pysql_examples.day_mapper import map_days
OutputDataSet = map_days(InputDataSet)
', 
@input_data_1 = N'SELECT * FROM DaysOfWeek',
@input_data_1_name = N'InputDataSet'
WITH RESULT SETS (("DayOfWeek" nvarchar(10) null,"DayOfWeekNumber" int null))

И вот (в качестве второго примера) простое приложение от Аликс Гамильтон на GitHub, которое выполняет запрос с BigQuery, просто используя функцию Python:

from google.cloud import bigquery

def query_stackoverflow():
    client = bigquery.Client()
    query_job = client.query("""SELECT CONCAT(
        'https://stackoverflow.com/questions/',
        CAST(id as STRING)) as url, 
        view_count FROM 'bigquery-public-data.stackoverflow.posts_questions'
        WHERE tags like '%google-bigquery%'
        ORDER BY view_count DESC
        LIMIT 10""")

    results = query_job.result()

    for row in results:
        print("{} : {} views".format(row.url, row.view_count))

if __name__ == '__main__':
    query_stackoverflow()

Надеюсь это поможет.

Ответ 2

У меня нет опыта делать это. Однако, основываясь на текущей способности BigQuery, может быть возможно запустить скомпилированную в javascript-функцию python или запустить скрипт python из интерпретатора Python на основе javascript. Здесь ссылка и ссылка.

Ответ 3

Python 3 Apache Beam + BigQuery Вот ключевой код Beam для чтения из BigQuery и записи в BigQuery:

with beam.Pipeline(RUNNER, options = opts) as p:
    (p 
      | 'read_bq' >> beam.io.Read(beam.io.BigQuerySource(query=query, use_standard_sql=True))
      | 'compute_fit' >> beam.FlatMap(compute_fit)
      | 'write_bq' >> beam.io.gcp.bigquery.WriteToBigQuery(
          'ch05eu.station_stats', schema='station_id:string,ag:FLOAT64,bg:FLOAT64,cg:FLOAT64')
    )

По сути, мы выполняем запрос к таблице BigQuery, запускаем Python-метод compute_fit и записываем вывод в таблицу BigQuery. Это мой метод compute_fit. Как видите, это просто простой код Python:

def compute_fit(row):
  from scipy import stats
  import numpy as np
  durations = row['duration_array']
  ag, bg, cg = stats.gamma.fit(durations)
  if np.isfinite(ag) and np.isfinite(bg) and np.isfinite(cg):
      result = {}
      result['station_id'] = str(row['start_station_id'])
      result['ag'] = ag
      result['bg'] = bg
      result['cg'] = cg
      yield result

Обязательно укажите пакеты Python, которые вам нужно установить на рабочих потока данных, в файле require.txt:

%%writefile requirements.txt
numpy
scipy

Наслаждайтесь! за дополнительной информацией вы можете обратиться к этому документу. Как запустить код Python в вашей таблице BigQuery