在 kedro 中将 csv 转换为镶木地板
Convert csv into parquet in kedro
我的 CSV 文件很大,无法放入内存,我需要将其转换为 .parquet 文件才能与 vaex 一起使用。
这是我的目录:
raw_data:
type: kedro.contrib.io.pyspark.SparkDataSet
filepath: data/01_raw/data.csv
file_format: csv
parquet_data:
type: ParquetLocalDataSet
filepath: data/02_intermediate/data.parquet
节点:
def convert_to_parquet(data: SparkDataSet) -> ParquetLocalDataSet:
return data.coalesce(1)
和管道:
def create_pipeline(**kwargs):
return Pipeline(
[
node(
func=convert_to_parquet,
inputs="raw_data",
outputs="parquet_data",
name="data_to_parquet",
),
]
)
但是如果我这样做 kedro run
我会收到此错误 kedro.io.core.DataSetError: Failed while saving data to data set ParquetLocalDataSet(engine=auto, filepath=data/02_intermediate/data.parquet, save_args={}).
'DataFrame' object has no attribute 'to_parquet'
我应该解决什么问题才能转换我的数据集?
您可以尝试以下方法。这在过去对我有用。
parquet_data:
type: kedro.contrib.io.pyspark.SparkDataSet
file_format: 'parquet'
filepath: data/02_intermediate/data.parquet
save_args:
我的 CSV 文件很大,无法放入内存,我需要将其转换为 .parquet 文件才能与 vaex 一起使用。
这是我的目录:
raw_data:
type: kedro.contrib.io.pyspark.SparkDataSet
filepath: data/01_raw/data.csv
file_format: csv
parquet_data:
type: ParquetLocalDataSet
filepath: data/02_intermediate/data.parquet
节点:
def convert_to_parquet(data: SparkDataSet) -> ParquetLocalDataSet:
return data.coalesce(1)
和管道:
def create_pipeline(**kwargs):
return Pipeline(
[
node(
func=convert_to_parquet,
inputs="raw_data",
outputs="parquet_data",
name="data_to_parquet",
),
]
)
但是如果我这样做 kedro run
我会收到此错误 kedro.io.core.DataSetError: Failed while saving data to data set ParquetLocalDataSet(engine=auto, filepath=data/02_intermediate/data.parquet, save_args={}).
'DataFrame' object has no attribute 'to_parquet'
我应该解决什么问题才能转换我的数据集?
您可以尝试以下方法。这在过去对我有用。
parquet_data:
type: kedro.contrib.io.pyspark.SparkDataSet
file_format: 'parquet'
filepath: data/02_intermediate/data.parquet
save_args: