将PySpark RDD上载到BigQuery

Question

我从BQ下载了一个表格到PySpark RDD，如下所示。我如何再次上传它？

dGSConfig = {
    'project_id': "project_id",
    'bucket': "bucket_id"
}
dBQConfig = {
    'gs_config': dGSConfig,
    'project_id': "project_id",
    'dataset_id': "dataset_id",
    'table_id': "table_id"
}

oSc = instantiate_pyspark()
rddData, lsHeadings = get_table_cloud(oSc, dBQConfig)  #rddData has a list-of-lists type format




def instantiate_pyspark():
    """instantiate the pyspark RDD stuff"""
    import pyspark

    oSc = pyspark.SparkContext()
    oHadoopConf = oSc._jsc.hadoopConfiguration()
    oHadoopConf.get("fs.gs.system.bucket")

    return oSc


def get_table_cloud(oSc, dBQConfig):
    """get a table from bigquery via google cloud storage
    Config format:
        dGSConfig = {'project_id': '', 'bucket':  ''}
        dBQConfig = {'project_id: '', 'dataset_id': '', 'table_id': ''}
    """
    dGSConfig = dBQConfig['gs_config']

    dConf = {
        "mapred.bq.project.id": dGSConfig['project_id'],
        "mapred.bq.gcs.bucket": dGSConfig['bucket'],
        "mapred.bq.input.project.id": dBQConfig['project_id'],
        "mapred.bq.input.dataset.id":dBQConfig['dataset_id'],
        "mapred.bq.input.table.id": dBQConfig['table_id']
    }

    rddDatasetRaw = oSc.newAPIHadoopRDD(
        "com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat",
        "org.apache.hadoop.io.LongWritable",
        "com.google.gson.JsonObject",
        conf=dConf
    )

    import json
    lsHeadings = json.loads(rddDatasetRaw.take(1)[0][1]).keys()

    rddDataset = (
        rddDatasetRaw
        .map(lambda t, json=json: json.loads(t[1]).values() )
    )

    return rddDataset, lsHeadings

Answer 1

我曾经用过的三种方法:

1） 创建本地csv，上传到google存储，单独的进程进入BigQuery：

llData = rddData.collect()


with open(sCsvPath, 'w') as f:
    import csv
    oWriter = csv.writer(f)
    for lData in llData:
        oWriter.writerow(lData)

import subprocess
lsCommand = ['gsutil', 'cp', sCsvPath, sGooglePath]
subprocess.check_output(lsCommand)

2） 使用Pandas直接上传到BigQuery：

import pandas as pd
dfData = pd.DataFrame(llData, columns=lsHeadings)

sProjectID = dBQConfig['sProjectID']
sTargetDataset = dBQConfig['sTargetDataset']
sTargetTable = dBQConfig['sTargetTable']

sTablePath = "{}.{}".format(sTargetDataset, sTargetTable)
dfData.to_gbq(sTablePath, sProjectID, if_exists='replace')

3） 使用 pyspark 将分布式结果直接保存到存储中：

#remove previous dir if exists
import subprocess
lsCommand = ['gsutil', 'rm', '-r', sGooglePath]
subprocess.check_output(lsCommand)

rddSave.saveAsTextFile(sGooglePath)

虽然这些都不是我最初想要的，但这是一种将结果直接上传到 BQ 的 PySpark 方式。

Answer 2

您可以导出到一些中间文件，然后将这些文件加载到BigQuery中。

这可能会有所帮助:如何将pyspark中的表格数据帧导出为csv？