我在大查询中有一个相当大的表(app. 9M 行),我想通过 Pandas 读取它。
我试过阅读和使用 [pd.read_gbq()][1]
功能,在小 table 上工作正常。
在大 table 上它会在 50 秒左右后卡住(日志显示 elapsed .. 50s
) - 没有给出错误或任何东西。
我的问题是如何使用 pd (块?)读取该表。任何有关扩大这些 bigquery 读取的约定都会有所帮助。
编辑/分辨率
添加到可汗的回答中,我最终实现了块,每次将 500,000 写入一个文件,然后将这些文件读取到数据帧,如下所示:
def download_gbq_table(self):
if not os.path.exists(self.tmp_dir):
os.makedirs(self.tmp_dir)
increment = 100000
intervals = list(range(0, self.table_size, 100000))
intervals.append(self.table_size - intervals[len(intervals)-1])
df = pd.DataFrame()
for offset in intervals:
query = f"select * from `<table_name>` limit {increment} offset {offset};"
logger.info(f"running query: {query}")
start_time = time.time()
tmp_df = pd.read_gbq(query,
project_id=self.connection_parameters['project_id'],
private_key=self.connection_parameters['service_account'],
dialect='standard'
)
df = pd.concat([df, tmp_df])
logger.info(f'time took: {str(round(time.time() - start_time, 2))}')
if len(df) % 500000 == 0:
df.to_csv(os.path.join(self.tmp_dir, f'df_{str(offset + increment)}.csv'))
df = pd.DataFrame()
def read_df_from_multi_csv(self):
all_files = glob.glob(os.path.join(self.tmp_dir, "df_*"))
df_list = []
for f in all_files:
start_time = time.time()
df_list.append(pd.read_csv(f))
logger.info(f'time took for reading {f}: {str(round(time.time() - start_time, 2))}')
return pd.concat((pd.read_csv(f) for f in all_files))
最佳答案
Pandas 的read_gbq
功能目前不提供chunksize
参数(即使它的相反 to_gbq
函数确实提供了一个 chunksize
参数)。
无论如何,您可以通过添加 LIMIT
来解决您的问题。和 OFFSET
从 BigQuery 迭代读取 SQL 查询内容。一些关于:
project_id = "xxxxxxxx"
increment=100000
chunks=range(0, 9000000, 100000)
chunks[-1]+=increment
intervals=[[chunks[i-1], chunks[i]+1] for i, e in enumerate(chunks) if i > 0]
query_str="select * from `mydataset.mytable` limit {end} offset {start};"
for start, end in intervals:
query = query_str.format(start=start, end=end)
df = pd.read_gbq(query, project_id)
#-- do stuff with your df here..
关于尝试从 bigquery 读取时,pandas 卡住了,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/54673272/