codewarnab opened a new pull request, #41186:
URL: https://github.com/apache/beam/pull/41186
Fixes #41081
## What this does
Beam YAML `ReadFromCsv` and `ReadFromJson` now accept `error_handling`:
```yaml
- type: ReadFromCsv
config:
path: gs://bucket/raw/*.csv
error_handling:
output: errors
```
Good records go to the main output. Bad records go to `errors`, and the
pipeline keeps running.
Each error row uses the standard YAML error format (`element`, `msg`,
`stack`). The `element` has:
- `path`: the file that the record came from
- `record`: the raw text of the bad record
A record is "bad" when:
- CSV: it has the wrong number of fields, or a value does not match the
column type.
- JSON: the line is not valid JSON, is not a JSON object, or a value does
not match the column type.
- Both: the line is not valid UTF-8.
**Without `error_handling`, nothing changes.** The code calls
`apache_beam.io.ReadFromCsv` / `ReadFromJson` exactly as before.
## Why this approach
The current readers parse each file chunk with pandas. When one row is bad,
pandas fails the whole chunk. We cannot tell which row was bad.
So, only when `error_handling` is set, this PR uses a separate read path in
`yaml_io.py`:
1. Read one sample from the first file (first 100 records) to get the column
names and types. This is the same idea as the current reader.
2. Read each file line by line, and group lines into records. A quoted CSV
field with newlines stays in one record.
3. Parse each record on its own with pandas, using the column types from
step 1.
4. If parsing fails, send the record to the error output.
I picked this option because:
- It does not touch the shared pandas reader
(`apache_beam/dataframe/io.py`). Other users of that reader are not affected.
- It only runs when a user asks for `error_handling`.
- The diff stays small and is easy to review.
**Other option:** change the pandas-based reader itself, so that it can
report bad rows. That is a bigger change. If maintainers prefer it, I am happy
to do that instead.
## Known limits
- Files are read one by one, not split into pieces. This is slower for very
large files.
- Column types come from the first 100 records of the first file. If a bad
value is in those 100 records, the column type may become string, and that
value is not reported as an error.
- `ReadFromJson` with `error_handling` supports newline-delimited JSON only.
This is already the YAML default.
## Testing
New tests in `yaml_io_test.py`:
- CSV: good rows, a quoted field with a newline, a wrong type, an extra
field, and invalid UTF-8.
- JSON: good lines, a broken line, and a line that is not an object.
- CSV without `error_handling` still fails on a bad row, as before.
I also ran the existing `yaml_io_test.py`, `yaml_transform_test.py` and
`yaml_provider_unit_test.py`. They all pass.
------------------------
Thank you for your contribution! Follow this checklist to help us
incorporate your contribution quickly and easily:
- [x] Mention the appropriate issue in your description (for example:
`addresses #123`), if applicable. This will automatically add a link to the
pull request in the issue. If you would like the issue to automatically close
on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [x] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual
Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more
tips on [how to make review process
smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
