Azure Data Lake Gen2 Data Receiver

The azure_data_lake_gen2 data receiver connects to an Azure Data Lake Gen2 instance and reads files stored on it.

Overview

This connection will make use of the Azure Data Lake Gen 2 ADLS API. (Also referred to as “Storage Account”) ADLS is built on top of blob storage and has some key differences/advantages over the blob API:

• ACL/POSIX permissions support

• Hierarchical namespaces (like a native FileSystem)

• Better storage retrieval performance

Example:

If we want to get all the CSV files in the some_directory in the some_container_1 storage container our Azure Data Lake instance, our configuration will look something like this:

{
    "data_receiver": [
        {
            "data_receiver_name": "my_adlv2_receiver",
            "protocol": "azure_data_lake_gen2",
            "poll_interval": 10,
            "connections": [
                {
                    "connection_name": "my_adlv2_connection",
                    "connection_str": "DefaultEndpointsProtocol=https;AccountName={};AccountKey={};EndpointSuffix=core.windows.net",
                    "storage_container_name": "some_container_1",
                    "data_directory": "some_directory"
                }
            ],
            "streams": [
                {
                    "asset": "some_asset_1",
                    "stream_type": "secret",
                    "file_filter": ["*.csv"],
                    "parser": "My_CSV_Parser"
                }
            ],
            "parsers": [
                {
                  "parser_name": "My_CSV_Parser",
                  "parser_type": "csv"
                }
            ]
        }
    ]
}

Configuration:

The azure_data_lake_gen2 data receiver uses a file receiver that connects to an Azure Data Lake Gen2 instance.

All file receivers support these required and optional properties:

• data_receiver_name: Unique name of the data receiver. This name will be used to track the progress state of the data stream.

• protocol: Protocol to be used.

• streams: List of input data streams to read from.

  • asset: Asset identifier

  • stream_type: Type of data stream

  • parser: Name of the parser used to convert the file. This has to match one of the parser_name values in the parsers section.

  • file_filter: List of files to filter on. Items can be regular expressions. You must select either file_filter or path_filter but not both.

  • path_filter: List of paths to filter on. Items can be regular expressions. You must select either file_filter or path_filter but not both.

• connections: A list of connection settings used to connect to the data source. Please refer to the connection settings below.

• process_files_alphabetically: true to process all of the files retrieved by the connection(s) in alphabetical order. By default, this option is false, so files are processed by last modified (or created) time with oldest files first.

• process_ordered_connections: If there are multiple connections and you want to process (and transmit) the files received from one connection before another, set option to true to process files based on the order of the connection(s). By default, this option is set to false. This option is usually related to the Kafka transmit because events within a Kafka topic need to be in order. For example, if one connection receives historical files and another connection receives “realtime” files, you’ll want to enable this setting and order the connections with the historical connection first, so that events in the Kafka topic are in chronological order. Please note that the order of the files per connection is maintained: if process_files_alphabetically is enabled, files will be parsed in alphabetical order; if disabled, files will be parsed in chronological order based on the last modified time.

• parsers: A list of parsers that can be used for the input streams. Each parser has the following default properties:

  • parser_name: A customizable name used by streams to identify the parser.

  • parser_type: Type of parser to apply (e.g. csv)

  • parser_version: Version of the parser type to apply.

  FactoryTX has a few built-in parsers available for you to use. Please refer to the Parsers Configurations section of the manual for more details about them.

• skip_parser_errors: Whether the receiver will allow parser errors when processing files. If True, parsing errors will be skipped and logged in a quarantine log file while the receiver continues processing. Otherwise if we encounter a parsing error, the receiver will halt until the file gets repaired. (Only works for csv files)

• archive_completed: true to keep a local copy of each file received from the remote server, or false if local copies should not be kept. If enabled, files will be archived until their total size increases above the amount specified in the max_archive_size_mb parameter.

• max_archive_size_mb: If archive_completed is true, delete the archive completed files once the total size (in mb) is greater than this value. A negative value means never delete any files. Defaults to -1.

• delete_completed: true if files should be deleted from the data directory after they have been received by FactoryTX, or false if files should never be deleted. Defaults to false to avoid accidentally losing data.

  • archive_completed and delete_completed are independent of each other. Archiving will create a copy of the file in a new directory, while deleting will remove the original file.

• read_in_progress_files: Whether to read files as soon as they are created, or wait for the upstream service to stop writing to the file before reading it.

• emit_file_metadata: Whether or not to inject additional columns into each record containing metadata about the file it came from. If this setting is enabled then every record will contain fields named file_name, file_path, and file_timestamp.

• poll_interval: Number of seconds to wait before fetching new data from the server.

• temporary_file_directory: Specify the directory to temporarily store files that have been downloaded from the connection(s). By default, the directory used is /tmp.

Connection Settings:

Required and optional properties that can be configured for an Azure Data Lake Gen2 connection:

• connection_name: Unique name for the connection.

• connection_str: connection string for connecting with azure data lake (In “Storage Accounts”, click on “Access Keys” on the sidebar to find.)

• storage_container_name: The storage_container_name to connect to

• data_directory: Path to a subdirectory containing files to fetch. If this option is not set, then all files in the store will be scanned recursively.

Please refer to Notes on ordered file processing for more details about file receiver options.