com.datastax.oss.dsbulk.connectors.commons

Class AbstractFileBasedConnector

java.lang.Object

com.datastax.oss.dsbulk.connectors.commons.AbstractFileBasedConnector

All Implemented Interfaces:

Connector, AutoCloseable

Direct Known Subclasses:

CSVConnector, JsonConnector

public abstract class AbstractFileBasedConnector
extends Object
implements Connector

A parent class for connectors that read from and write to text-based files.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static interface	AbstractFileBasedConnector.RecordReader A reader for Records.
protected static interface	AbstractFileBasedConnector.RecordWriter A writer for Records.

Field Summary

Fields

Modifier and Type	Field and Description
protected String	compression
protected static String	COMPRESSION
protected Charset	encoding
protected static String	ENCODING
protected static String	FILE_NAME_FORMAT
protected static String	FILE_NAME_PATTERN
protected AtomicInteger	fileCounter
protected String	fileNameFormat
protected List<URL>	files
protected static String	MAX_CONCURRENT_FILES
protected static String	MAX_RECORDS
protected int	maxConcurrentFiles
protected long	maxRecords
protected AtomicInteger	nextWriterIndex
protected String	pattern
protected boolean	read
protected boolean	recursive
protected static String	RECURSIVE
protected int	resourceCount
protected boolean	retainRecordSources
protected List<Path>	roots
protected AbstractFileBasedConnector.RecordWriter	singleWriter
protected static String	SKIP_RECORDS
protected long	skipRecords
protected static String	URL
protected static String	URLFILE
protected List<URL>	urls
protected Deque<AbstractFileBasedConnector.RecordWriter>	writers

Constructor Summary

Constructors

Constructor and Description
AbstractFileBasedConnector()

Method Summary

Modifier and Type	Method and Description
protected Flux<Record>	applyPerFileLimits(Flux<Record> records) Applies per-file limits to a stream of records coming from readSingleFile(java.net.URL).
void	close() Closes the connector.
void	configure(Config settings, boolean read, boolean retainRecordSources) Configures the connector.
protected abstract String	getConnectorName() Returns the connector name, e.g.
protected URL	getOrCreateDestinationURL() Returns the URL that the connector should write to.
void	init() Initializes the connector.
protected boolean	isDataSizeSamplingAvailable() Checks whether it is safe to perform data size sampling on this connector's data source.
protected List<URL>	loadURLs(Config settings) Validates and computes the list of URLs to be read or written.
protected abstract AbstractFileBasedConnector.RecordReader	newSingleFileReader(URL url) Returns a new AbstractFileBasedConnector.RecordReader instance; cannot be null.
protected abstract AbstractFileBasedConnector.RecordWriter	newSingleFileWriter() Returns a new AbstractFileBasedConnector.RecordWriter instance; cannot be null.
protected void	processURLsForRead() Inspects the list or URLs as loaded by loadURLs(Config) and determines exactly what files and folders need to be read.
protected void	processURLsForWrite() Inspects the list or URLs as loaded by loadURLs(Config) and determines if the connector should write to a single file, or to a directory of files.
Publisher<Publisher<Record>>	read() Reads all records from the datasource in a flow of flows that can be consumed in parallel.
int	readConcurrency() Returns the desired read concurrency, that is, how many resources are expected to be read in parallel.
protected Flux<Record>	readSingleFile(URL url) Reads a single text file accessible through the given URL.
protected Flux<URL>	scanRootDirectory(Path root) Scans a directory for readable files and returns the files found as a stream.
Function<Publisher<Record>,Publisher<Record>>	write() Returns a function that handles writing records to the datasource.
int	writeConcurrency() Returns the desired write concurrency, that is, how many resources are expected to be written in parallel.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface com.datastax.oss.dsbulk.connectors.api.Connector

getRecordMetadata, supports

Field Detail

URL

protected static final String URL

See Also:

Constant Field Values

URLFILE

protected static final String URLFILE

See Also:

Constant Field Values

COMPRESSION

protected static final String COMPRESSION

See Also:

Constant Field Values

ENCODING

protected static final String ENCODING

See Also:

Constant Field Values

FILE_NAME_PATTERN

protected static final String FILE_NAME_PATTERN

See Also:

Constant Field Values

SKIP_RECORDS

protected static final String SKIP_RECORDS

See Also:

Constant Field Values

MAX_RECORDS

protected static final String MAX_RECORDS

See Also:

Constant Field Values

MAX_CONCURRENT_FILES

protected static final String MAX_CONCURRENT_FILES

See Also:

Constant Field Values

RECURSIVE

protected static final String RECURSIVE

See Also:

Constant Field Values

FILE_NAME_FORMAT

protected static final String FILE_NAME_FORMAT

See Also:

Constant Field Values

read

protected boolean read

retainRecordSources

protected boolean retainRecordSources

urls

protected List<URL> urls

roots

protected List<Path> roots

files

protected List<URL> files

encoding

protected Charset encoding

compression

protected String compression

fileNameFormat

protected String fileNameFormat

recursive

protected boolean recursive

pattern

protected String pattern

skipRecords

protected long skipRecords

maxRecords

protected long maxRecords

resourceCount

protected int resourceCount

maxConcurrentFiles

protected int maxConcurrentFiles

writers

protected Deque<AbstractFileBasedConnector.RecordWriter> writers

singleWriter

protected AbstractFileBasedConnector.RecordWriter singleWriter

fileCounter

protected AtomicInteger fileCounter

nextWriterIndex

protected AtomicInteger nextWriterIndex

Constructor Detail

AbstractFileBasedConnector

public AbstractFileBasedConnector()

Method Detail

readConcurrency

public int readConcurrency()

Description copied from interface: Connector

Returns the desired read concurrency, that is, how many resources are expected to be read in parallel.

The workflow runner is guaranteed to never consume more inner flows in parallel than readConcurrency.

Depending on the read concurrency, and whether it is greater than the number of cores or not, the runner will try to assign one thread to each resource, thus eliminating any context switch, from the resource to read until the final statements to execute.

This method should only be called after the connector is properly configured and initialized.

Specified by:

readConcurrency in interface Connector

Returns:

the desired read concurrency; must be strictly positive, that is, greater than zero.

writeConcurrency

public int writeConcurrency()

Description copied from interface: Connector

Returns the desired write concurrency, that is, how many resources are expected to be written in parallel.

The workflow runner is guaranteed to never invoke the Connector.write() function by more than writeConcurrency threads concurrently.

Depending on the write concurrency, and whether it is greater than the number of cores or not, the runner will decide to assign one thread to each resource, thus eliminating any context switch, from the decoded rows until the final resource being written.

This method should only be called after the connector is properly configured and initialized.

Specified by:

writeConcurrency in interface Connector

Returns:

the desired write concurrency; must be strictly positive, that is, greater than zero.

configure

public void configure(@NonNull
                      Config settings,
                      boolean read,
                      boolean retainRecordSources)

Description copied from interface: Connector

Configures the connector.

Specified by:

configure in interface Connector

Parameters:

settings - the connector settings.

read - whether the connector should be configured for reading or writing.

retainRecordSources - whether the connector should retain sources when emitting records; only applicable when the connector is being configured for reads.

init

public void init()
          throws URISyntaxException,
                 IOException

Description copied from interface: Connector

Initializes the connector.

This method should only be called after the connector is properly configured.

Specified by:

init in interface Connector

Throws:

URISyntaxException

IOException

read

@NonNull
public Publisher<Publisher<Record>> read()

Description copied from interface: Connector

Reads all records from the datasource in a flow of flows that can be consumed in parallel.

The inner flows are guaranteed to be consumed with a parallelism no greater than the read concurrency.

If the underlying datasource cannot split its records into multiple flows in any efficient manner, then this method should return a publisher of one single inner publisher.

This method should only be called after the connector is properly configured and initialized.

Specified by:

read in interface Connector

Returns:

a Publisher of records read from the datasource, grouped by resources.

write

@NonNull
public Function<Publisher<Record>,Publisher<Record>> write()

Description copied from interface: Connector

Returns a function that handles writing records to the datasource.

This method should only be called after the connector is properly configured and initialized.

Specified by:

write in interface Connector

Returns:

A transforming Function that writes records from the upstream flow to the datasource, then emits the records written to downstream subscribers.

close

public void close()

Description copied from interface: Connector

Closes the connector.

Specified by:

close in interface Connector

Specified by:

close in interface AutoCloseable

getConnectorName

@NonNull
protected abstract String getConnectorName()

Returns the connector name, e.g. "csv" for the CSV connector. This should match the programmatic connector name and also its configuration namespace; that is, if this method returns "foo", then the connector configuration namespace is expected to be connector.foo.

readSingleFile

@NonNull
protected Flux<Record> readSingleFile(@NonNull
                                               URL url)

Reads a single text file accessible through the given URL. Used during the data reading phase.

Implementors should not care about maxRecords and skipRecords, these will be applied later on, see applyPerFileLimits(Flux).

Parameters:

url - The URL to read; must not be null; must be accessible and readable (but not necessarily hosted on the local filesystem).

Returns:

A stream of Records; never null but may be empty.

newSingleFileReader

@NonNull
protected abstract AbstractFileBasedConnector.RecordReader newSingleFileReader(@NonNull
                                                                                        URL url)
                                                                                 throws IOException

Returns a new AbstractFileBasedConnector.RecordReader instance; cannot be null. Only used when reading. Each invocation of this method is expected to return a newly-allocated instance. The reader is expected to be initialized already, and ready to emit its first record. It is possible to throw IOException if the reader cannot be initialized.

Throws:

IOException

newSingleFileWriter

@NonNull
protected abstract AbstractFileBasedConnector.RecordWriter newSingleFileWriter()

Returns a new AbstractFileBasedConnector.RecordWriter instance; cannot be null. Only used when writing. Each invocation of this method is expected to return a newly-allocated instance.

loadURLs

@NonNull
protected List<URL> loadURLs(@NonNull
                                      Config settings)

Validates and computes the list of URLs to be read or written. This method honors both the URLFILE and URL configuration options.

Expected to be called during the configuration phase.

processURLsForRead

protected void processURLsForRead()
                           throws URISyntaxException,
                                  IOException

Inspects the list or URLs as loaded by loadURLs(Config) and determines exactly what files and folders need to be read.

Should be called at the beginning of the initialization process, but only when reading, never when writing.

This method expects that loadURLs(Config) has been previously called.

Throws:

URISyntaxException

IOException

processURLsForWrite

protected void processURLsForWrite()
                            throws URISyntaxException,
                                   IOException

Inspects the list or URLs as loaded by loadURLs(Config) and determines if the connector should write to a single file, or to a directory of files.

Should be called at the beginning of the initialization process, but only when writing, never when reading.

This method expects that loadURLs(Config) has been previously called, and also expects exactly one URL to be present, which can be either a directory or a file.

Throws:

URISyntaxException

IOException

scanRootDirectory

@NonNull
protected Flux<URL> scanRootDirectory(@NonNull
                                               Path root)

Scans a directory for readable files and returns the files found as a stream. Only used when reading, never when writing. Normally used as part of the actual data reading phase.

applyPerFileLimits

@NonNull
protected Flux<Record> applyPerFileLimits(@NonNull
                                                   Flux<Record> records)

Applies per-file limits to a stream of records coming from readSingleFile(java.net.URL).

getOrCreateDestinationURL

@NonNull
protected URL getOrCreateDestinationURL()

Returns the URL that the connector should write to. Not used for reads.

This can be either a single file or a directory of files. If the former, each invocation of this method will return the same URL; if the latter, each invocation of this method will generate a new URL inside the directory, with a unique file name.

isDataSizeSamplingAvailable

protected boolean isDataSizeSamplingAvailable()

Checks whether it is safe to perform data size sampling on this connector's data source. Data size sampling is usually not safe if the data can only be streamed once.

This implementation simply checks that none of the urls to read is reading from standard input, since standard input is not rewindable. Any other URL protocol is considered safe.

Returns:

true if it is safe to perform data size sampling, false otherwise.