apache / apache/datafusion

Writing to CSV create multiple partition file but other files are empty except partition Zero

Open
#5,467 1 comment 0 reactions 0 assignees View on GitHub
bug
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

**Describe the bug**
Hi, I was trying a data processing pipeline. Here is the full code of the pipeline. While writing the result into CSV by default it creates several csv partition files but all the data are in the first file and other files are empty. How can I control the partition behavior?

Here are the file created by the program:

```
⇒ ls data/processed/diabetes
part-0.csv part-1.csv part-2.csv part-3.csv part-4.csv part-5.csv part-6.csv part-7.csv
```

**To Reproduce**
Here is the full code of the processing:

```
use datafusion::error::Result;
use datafusion::prelude::*;
use log::info;
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use std::fs;
use env_logger;

#[tokio::main]
async fn main() -> Result<()>{
env_logger::init();
let df = read_diabetes_dataset().await;
let df = filter_zero_values(df).await;
df.clone().limit(0, Some(5))?.show().await?;
run_pipeline(df).await?;
Ok(())
}

async fn read_diabetes_dataset() -> DataFrame{
let path = "data/raw/diabetes.csv";
let schema = Schema::new(vec![
Field::new("Pregnancies", DataType::Int32, false),
Field::new("Glucose", DataType::Int32, false),
Field::new("BloodPressure", DataType::Int32, false),
Field::new("SkinThickness", DataType::Int32, false),
Field::new("Insulin", DataType::Int32, false),
Field::new("BMI", DataType::Float32, false),
Field::new("DiabetesPedigreeFunction", DataType::Float32, false),
Field::new("Age", DataType::Int32, false),
Field::new("Outcome", DataType::Int8, false),
]);
let csv_read_optoins = CsvReadOptions{schema: Some(&schema), ..Default::default()};
let ctx = SessionContext::new();
let df = ctx.read_csv(path.to_string(), csv_read_optoins).await.unwrap();
info!("File read successfully into {}", path);
df
}

async fn filter_zero_values(df: DataFrame) -> DataFrame {
let df = df
.filter(col("Glucose").gt(lit(0))).unwrap()
.filter(col("BloodPressure").gt(lit(0))).unwrap()
.filter(col("SkinThickness").gt(lit(0))).unwrap()
.filter(col("Insulin").gt(lit(0))).unwrap()
.filter(col("BMI").gt(lit(0.0))).unwrap()
.filter(col("Age").gt(lit(0))).unwrap()
;
df
}

async fn run_pipeline(df: DataFrame) -> Result<()> {
let write_path = "data/processed/diabetes";
match fs::remove_dir_all(write_path) {
Ok(_) => info!("File already existed while writting and removed!"),
Err(_) => info!("No file found while cleaning the write path!")
};
info!("Row count before processing.");
let result = df.clone().aggregate(vec![], vec![count(col("Pregnancies"))])?;
result.show_limit(1).await?;
let df = filter_zero_values(df).await;
let result = df.clone().aggregate(vec![], vec![count(col("Pregnancies"))])?;
info!("Row count after processing.");
result.show_limit(1).await?;
df.write_csv(write_path).await?;
info!("File written successfully into {}", write_path);
Ok(())
}
```
I did check the repartitioning API [doc](https://docs.rs/datafusion/latest/datafusion/physical_plan/enum.Partitioning.html). But there is no specific example there. The data can be download from Kaggle [diabetes data set](https://www.kaggle.com/code/adamhertelendi/diabetes-ds-data-analysis).

Dep. list:
```
[dependencies]
tokio = "1.26.0"
datafusion = "19.0.0"
log = "0.4.0"
env_logger = "0.9.0"
```

**Expected behavior**
The write should create only one partition as long I do not say it to create more partitions or I do not re-partition the data frame.

Contributor guide

Open the contributing guide

Research direction

Start by running the provided Rust pipeline with DataFusion 19.0.0 and inspect the behavior of df.write_csv(write_path). Read the linked Partitioning API documentation and the CSV writing entry point to determine why empty partition files are created. Done means the observed partition behavior is explained and the output matches the stated expectation or the expected behavior is clarified.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
data-engineering
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
42/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.