将 Google BigQuery 中一个表中的 XML 数据转换为同一表中另一列中的 JSON 数据

Posted 2023-03-25

【中文标题】将 Google BigQuery 中一个表中的 XML 数据转换为同一表中另一列中的 JSON 数据

【英文标题】：Convert XML data from one table in Google BigQuery to JSON data in another column in the same table

【发布时间】：2019-11-11 12:11:52

【问题描述】：

我在 Google BigQuery 中有下表（这里只显示了几行）：

id     loaded_date     data
1      2019-10-25      <collection><row><field name="Item Key" type="text" value="Haircolour - Avstemming kunder - OMT" /><field name="Created" type="datetime" value="2019-10-25 17:35:17Z" /><field name="Type" type="text" value="Session Provisioning Failure" /></row></collection>
2      2019-10-25      <collection><row><field name="Item Key" type="text" value="Haircolour - Avstemming kunder - OMT" /><field name="Created" type="datetime" value="2019-10-25 17:51:32Z" /><field name="Type" type="text" value="Session Provisioning Failure" /></row></collection>
3      2019-02-23      <collection><row><field name="Item Key" type="text" value="Haircolour - Hent klienter til kø" /><field name="Last Generation Time" type="datetime" value="2019-02-23 11:00:36Z" /><field name="Priority" type="number" value="-3" /></row></collection>

我的 data 列是 XML 格式。我想在此表中添加第四列，例如名为 data_json 的列，其中包含与 data 列中相同的数据，但采用 JSON 格式。

这意味着我希望得到以下结果：

id     loaded_date     data                    data_json
1      2019-10-25      Same data as before     "collection": "row": "field": ["-name": "Item Key","-type": "text","-value": "Haircolour - Avstemming kunder - OMT","-name": "Created","-type": "datetime","-value": "2019-10-25 17:35:17Z","-name": "Type","-type": "text","-value": "Session Provisioning Failure"]
2      2019-10-25      Same data as before     "collection": "row": "field": ["-name": "Item Key","-type": "text","-value": "Haircolour - Avstemming kunder - OMT","-name": "Created","-type": "datetime","-value": "2019-10-25 17:51:32Z","-name": "Type","-type": "text","-value": "Session Provisioning Failure"]
3      2019-02-23      Same data as before     "collection": "row": "field": ["-name": "Item Key","-type": "text","-value": "Haircolour - Hent klienter til kø","-name": "Last Generation Time","-type": "datetime","-value": "2019-02-23 11:00:36Z","-name": "Priority","-type": "number","-value": "-3"]

有没有办法在 BIgquery 中直接使用 SQL 或使用 Python 来做到这一点？

谢谢

【问题讨论】：

AFAIK，您不能直接在 SQL 中使用任何函数执行此操作。最简单的方法是编写 javascript UDF 并在其中使用 XML -> JSON 库。请参阅此处开始：***.com/questions/48954109/…

谢谢@GrahamPolley。这非常有用。

【参考方案1】：

要更新 BigQuery 中的数据，您可以查看 Data Manipulation Language，但请注意它有自己的配额。在您的情况下，我会考虑从现有表创建一个新表，并在 Python 中处理 XML 字段以将其解析为 JSON 格式。

我已经使用适用于 Python 的 Google Cloud 客户端库重现了我的工作流程，并且它与下面的附加代码一起正常工作。此代码的工作原理如下：

将表格 CSV 文件导出到 GCS 存储桶 将 CSV 文件从 GCS 存储桶下载到您的计算机 将列附加到名为“JSON_data”的输入 DataFrame 将 XML 列“数据”解析为“JSON_data”列中的 JSON 格式 使用新数据创建一个新的 BigQuery 表

为了创建 BigQuery 表，我关注了this *** 线程。

您必须设置自己的变量（bucket_name、project、dataset_id、table_id、location）。请记住让您的 GCS 存储桶与您的 BigQuery 数据集位于同一区域。

import xmltodict, json
from google.cloud import bigquery
from google.cloud import storage
import pandas as pd


#Define bigquery Client
client = bigquery.Client()

#Extract job
bucket_name = <YOUR_BUCKET_NAME>
project = <YOUR_PROJECT_ID>
dataset_id = <YOUR_DATASET_ID>
table_id = <YOUR_TABLE_ID>
location = <YOUR_TABLE_LOCATION>


def export_dataset(bucket_name, dataset_id, project, table_id):

    destination_uri = "gs:///".format(bucket_name, "bq_table.csv")
    dataset_ref = client.dataset(dataset_id, project=project)
    table_ref = dataset_ref.table(table_id)

    extract_job = client.extract_table(
        table_ref,
        destination_uri,
        # Location must match that of the source table.
        location=location,
    )  # API request
    extract_job.result()  # Waits for job to complete.

    print(
        "Exported :. to ".format(project, dataset_id, table_id, 
destination_uri)
    )


#Execute export job    
export_dataset(bucket_name, dataset_id, project, table_id)


#--------------------------------------------

#Retrieve CSV file from GCS bucket
source_blob_name = "bq_table.csv"
destination_file_name = "bq_table.csv"

def download_blob(bucket_name, source_blob_name, destination_file_name):
    """Downloads a blob from the bucket."""
    storage_client = storage.Client()
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(source_blob_name)

    blob.download_to_filename(destination_file_name)

    print('Blob  downloaded to .'.format(
        source_blob_name,
        destination_file_name))

#Download CSV from bucket
download_blob(bucket_name, source_blob_name, destination_file_name)

#--------------------------------------------

#Declare XML column name
XML_col = 'data' 

#Read CSV as Pandas DF
df = pd.read_csv('bq_table.csv')
#Append JSON_data column
df['JSON_data'] = ''
#Transform XML and save in Array
JSON_arr = [json.dumps(xmltodict.parse(df[XML_col].values[i])) for i in 
 range(len(df[XML_col]))]
#Set transformed data to column JSON_data
df.loc[:,'JSON_data'] = JSON_arr
#df to CSV - Generete output file
df.to_csv('new_data.csv', index=False, sep=',')

#----------------------------------------------


#Now we will create the new table with the new CSV 
csv_path='gs:///new_data.csv'.format(bucket_name)
new_table='new_table'


#Define schema for table
schema = [
        bigquery.SchemaField("id", "INTEGER"),
        bigquery.SchemaField("loaded_date", "DATE"),
        bigquery.SchemaField("JSON_data", "STRING"),   
    ]

#https://***.com/questions/44947369/load-the-csv-file-into-big-query-auto- 
detect-schema-using-python-api
def insertTable(datasetName, tableName, csvFilePath, schema=None):
    """
    This function creates a table in given dataset in our default project
    and inserts the data given via a csv file.

    :param datasetName: The name of the dataset to be created
    :param tableName: The name of the dataset in which the table needs to be created
    :param csvFilePath: The path of the file to be inserted
    :param schema: The schema of the table to be created
    :return: returns nothing
    """

    csv_file = open(csvFilePath, 'rb')

    dataset_ref = client.dataset(datasetName)        
    from google.cloud.bigquery import Dataset
   dataset = Dataset(dataset_ref)

    table_ref = dataset.table(tableName)
    if schema is not None:
        table = bigquery.Table(table_ref,schema)
    else:
        table = bigquery.Table(table_ref)

    try:
        client.delete_table(table)
    except:
        pass

    table = client.create_table(table)

    from google.cloud.bigquery import LoadJobConfig        
    job_config = LoadJobConfig()
    table_ref = dataset.table(tableName)
    job_config.source_format = 'CSV'
    job_config.skip_leading_rows = 1
    job_config.autodetect = True
    job = client.load_table_from_file(
        csv_file, table_ref, job_config=job_config)
    job.result()

insertTable(dataset_id, new_table, 'new_data.csv', schema)

请告诉我这是否适合您。

---

【讨论】：

谢谢华金。我今天无法尝试，但我明天会尝试，并会告诉你。非常感谢。干杯。