LocalFileReceiver

class factorytx.receivers.localfile.LocalFileReceiver(dataflow: DataflowGraph, config: dict, root_config: dict)

The localfile data receiver fetches and reads files on the local system.

Example:

If we want to retrieve data from a local CSV file in /data/test-data/multiple-files, our configuration will look something like this:

{
    "data_receiver": [
        {
            "data_receiver_name": "my_localfile_receiver",
            "protocol": "localfile",
            "delete_completed": true,
            "poll_interval": 4,
            "connections": [
                {
                    "data_directory": "/data/test-data/multiple-files",
                    "connection_name": "my_connection"
                }
            ],
            "streams": [
                {
                    "asset": "Station_CSV_3",
                    "stream_type": "cycle",
                    "file_filter": ["station3.csv"],
                    "parser": "My_CSV_Parser"
                }
            ],
            "parsers": [
                {
                  "parser_name": "My_CSV_Parser",
                  "parser_type": "csv"
                }
            ]
        }
    ]
}

Configuration:

The localfile data receiver uses a file receiver that connects to the local system.

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.

• deduplicate_rows: Whether or not to deduplicate rows from processing files. This is done by hashing each row and saving it compare against later if the file is modified. If the file is modified, only the new rows will be processed. This should be used if the same file is expected to be re-processed multiple times and it will be regularly modified, rather than appended to. All data in each row must be a hashable type (no lists, dicts, etc.) File hash stores will be in the receiver’s component directory in the “row_hash_stores” directory.

• deduplicate_row_retention_time_days: The number of days to keep row hashes for deduplication (default 90).

Connection Settings:

Required and optional properties that can be configured for a local file connection:

• connection_name: Unique name for the connection.

• data_directory: Path to a directory on the local filesystem containing input files.

• recursive: When enabled, the localfile connection will recursively search for files in subdirectories of the data_directory. By default, this setting is set to true.