将 PySpark RDD 上传到 BigQuery答案

【问题标题】：Upload PySpark RDD into BigQuery将 PySpark RDD 上传到 BigQuery
【发布时间】：2016-08-15 05:09:17
【问题描述】：

我从 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

【问题讨论】：

标签： python python-2.7 google-bigquery pyspark

【解决方案1】：

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

这可能会有所帮助：how to export a table dataframe in pyspark to csv?

【讨论】：

【解决方案2】：

我在某个时候使用的 3 种方法：

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 方式。

【讨论】：