Refactor BQ to expose all beam's configurations #5456
Draft
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import org.apache.beam.sdk.values.{PCollection, PCollectionTuple, TupleTag} | ||
|
|
||
| /** | ||
| * A sink for error records. |
There was a problem hiding this comment.
bit more explanation on error records could be helpful, maybe:
Suggested change
| * A sink for error records. | |
| * A sink for error records. | |
| * | |
| * An error record is produced by certain PTransforms that catch processing exceptions and transform the resulting (element, exception) pair into a [[BadRecord]] instance. | |
| * When an ErrorSink is configured (via ScioContext#errorSink), these BadRecords can be accessed as an SCollection by invoking the ErrorSink#sink method. | |
| * An ErrorSink is useful if you'd like to set up special handling of exceptions (incrementing Counters, logging the exceptions in a database, etc). |
| * Once the [[sink]] is materialized, the [[handler]] must not be used anymore. | ||
| */ | ||
| sealed trait ErrorSink { | ||
| def handler: ErrorHandler[BadRecord, _] |
There was a problem hiding this comment.
could def handler be private[scio]? not sure when a user would need to access this
There was a problem hiding this comment.
This is the API exposed by beam. As mentioned in the description we do not pass the ErrorSink directly.
sc.bigQueryStorageFormat[MyType](
table,
format,
errorHandler = errorSink.handler
)I was thinking of adding to the ScioContext a beam java like API too
def registerBadRecordErrorHandler[T](handler: PTransform[PCollection[BadRecord], T] sinkTransform): BadRecordErrorHandler[OutputT]
RustedBones
force-pushed
the
bq-factorization
branch
from
f9c02ac to
2d66440
Compare
RustedBones
force-pushed
the
bq-factorization
branch
from
2d66440 to
cfd83f4
Compare
RustedBones
force-pushed
the
bq-factorization
branch
from
cfd83f4 to
37b545f
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Here are the main changes:
the BQ
Tablesource has a single normalized definition, with multiple constrictors (form string spec orTableReference). It nows includes an optionalTable.Filterthat can be used is the storage read API to project and filter.read API changes with
FormatAPI take aBigqueryIO.Formatobject allowing to convert either fromGenericRecord(this should be prefered) orTableRowStorageApi allow to pass anErrorHandler(Fix Add ErrorHandling in BigQuery #5530). In order to preserve a flat structureScioContext.errorSink(): ErrorSinkhas been added. This allow to do the followingThe
handlercan be passed to multiple IOs beforesinkis materialized. The sink will flatten the errors from the IOs.