Dataproc의 BigQuery 및 Pyspark

Question

BigQuery에 FPgrowth 알고리즘을 쿼리하고 구현하려는 테이블이 있습니다. dataproc 클러스터의 VM 인스턴스를 사용하여 pyspark 셸에서 먼저 시도하고 싶습니다.

나는 pyspark를 사용하여 BQ의 테이블을 직접 질의하는 방법을 찾고있다. FPGrowth를 구현하기 위해 결과 쿼리 데이터를 사용하고 싶습니다 (이미 익숙합니다).

Answer 1

Dataproc에는 docs에서 볼 수있는 BigQuery를 쿼리하는 데 필요한 커넥터가 이미 있습니다. 문서에서

코드 샘플 :

import pyspark 
from pyspark.sql import SQLContext 

sc = pyspark.SparkContext() 

# Use the Google Cloud Storage bucket for temporary BigQuery export data used 
# by the InputFormat. This assumes the Google Cloud Storage connector for 
# Hadoop is configured. 
bucket = sc._jsc.hadoopConfiguration().get('fs.gs.system.bucket') 
project = sc._jsc.hadoopConfiguration().get('fs.gs.project.id') 
input_directory = 'gs://{}/hadoop/tmp/bigquery/pyspark_input'.format(bucket) 

conf = { 
    # Input Parameters. 
    'mapred.bq.project.id': project, 
    'mapred.bq.gcs.bucket': bucket, 
    'mapred.bq.temp.gcs.path': input_directory, 
    'mapred.bq.input.project.id': 'publicdata', 
    'mapred.bq.input.dataset.id': 'samples', 
    'mapred.bq.input.table.id': 'shakespeare', 
} 

# Output Parameters. 
output_dataset = 'wordcount_dataset' 
output_table = 'wordcount_output' 

# Load data in from BigQuery. 
table_data = sc.newAPIHadoopRDD(
    'com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat', 
    'org.apache.hadoop.io.LongWritable', 
    'com.google.gson.JsonObject', 
    conf=conf) 

내가 또한 추천하는 것은 설치 Jupyter service와 통해 Dataproc 클러스터를 만드는 것입니다. 이렇게하면 FPgrowth를 구현하는 방법이나 궁극적으로 시도하려는 다른 아이디어를 즉석에서 테스트 할 수 있습니다.