Apache Beam 能否检测 Parquet 文件(如 Spark 和 Pandas)的模式(列名)?
Can Apache Beam detect the schema (column names) of a Parquet file like Spark and Pandas?
我是 Apache Beam 的新手,我来自 API 如此丰富的 Spark 世界。
如何使用 Apache Beam 获取 Parquet 文件的架构?否则我会在内存中加载数据,因为有时它可能会很大,我只对了解列以及可选的列类型感兴趣。
语言是Python。
存储系统是Google Cloud Storage,Apache Beam作业必须是Dataflow中的运行。
仅供参考,我已按照 sof 中的建议尝试了以下操作:
from pyarrow.parquet import ParquetFile
ParquetFile(source).metadata
首先,当我给它一个 gs://..
路径时它不起作用,给我这个错误:error: No such file or directory
然后我在我的机器上尝试了一个本地文件,我稍微改变了代码:
from pyarrow.parquet import ParquetFile
ParquetFile(source).metadata.schema
这样我就可以得到列 :
<pyarrow._parquet.ParquetSchema object at 0x10927cfd0>
name: BYTE_ARRAY
age: INT64
hobbies: BYTE_ARRAY String
但是在我看来,这个解决方案需要我将这个文件放到本地(Dataflow 服务器的??)并且它不使用 Apache Beam。
任何(更好的)解决方案?
谢谢!
我很高兴在阅读 apache_beam.io.parquetio
的代码源后想出一个手工解决方案 :
import pyarrow.parquet as pq
from apache_beam.io.parquetio import _ParquetSource
import os
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = '<json_key_path>'
ps = _ParquetSource("", None, None, None) # file_pattern, min_bundle_size, validate, columns
with ps.open_file("<GCS_path_of_parquet_file>") as f:
pf = pq.ParquetFile(f)
print(pf.metadata.schema)
我是 Apache Beam 的新手,我来自 API 如此丰富的 Spark 世界。
如何使用 Apache Beam 获取 Parquet 文件的架构?否则我会在内存中加载数据,因为有时它可能会很大,我只对了解列以及可选的列类型感兴趣。
语言是Python。
存储系统是Google Cloud Storage,Apache Beam作业必须是Dataflow中的运行。
仅供参考,我已按照 sof 中的建议尝试了以下操作:
from pyarrow.parquet import ParquetFile
ParquetFile(source).metadata
首先,当我给它一个 gs://..
路径时它不起作用,给我这个错误:error: No such file or directory
然后我在我的机器上尝试了一个本地文件,我稍微改变了代码:
from pyarrow.parquet import ParquetFile
ParquetFile(source).metadata.schema
这样我就可以得到列 :
<pyarrow._parquet.ParquetSchema object at 0x10927cfd0>
name: BYTE_ARRAY
age: INT64
hobbies: BYTE_ARRAY String
但是在我看来,这个解决方案需要我将这个文件放到本地(Dataflow 服务器的??)并且它不使用 Apache Beam。
任何(更好的)解决方案?
谢谢!
我很高兴在阅读 apache_beam.io.parquetio
的代码源后想出一个手工解决方案 :
import pyarrow.parquet as pq
from apache_beam.io.parquetio import _ParquetSource
import os
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = '<json_key_path>'
ps = _ParquetSource("", None, None, None) # file_pattern, min_bundle_size, validate, columns
with ps.open_file("<GCS_path_of_parquet_file>") as f:
pf = pq.ParquetFile(f)
print(pf.metadata.schema)