CSV::from_file does not work with remote URLs

[shadowhand]

I want to be able to extract documents stored on S3, so I have registered the stream wrapper:

$client = $sdk->createS3();
$client->registerStreamWrapper();

And then I attempt to load the file:

(new Flow())
    ->extract(CSV::from_file(
        file_name: 's3://my-bucket-name/test.csv',
        header_offset: 0,
    ))
    // ...
    ->run();

But the following exception is thrown:

 [Flow\ETL\Exception\InvalidArgumentException]                                                    
  File s3://my-bucket-name/test.csv not found.

I think the check for file existence should be modified to allow remote URLs:

if (! (str_contains($file_name, needle: '://') || is_file($file_name))) { /* throw exception */ }

OR add another method to handle resources, eg:

$extractor = CSV::from_resource(
    resource: fopen('s3://my-bucket-name/test.csv'),
    header_offset: 0,
);

[norberttech]

This is literally one of the last problems that need to be solved before we can release version 0.1.0.

Read/Write Streams

Long story short is that at this point all adapters are expecting files (except log, elasticsearch and doctrine dbal adapters).
What you suggested makes perfect sense but it needs to be done in a way that would satisfy all adapters.

passing resource:

$extractor = CSV::from_resource(
    resource: fopen('s3://my-bucket-name/test.csv'),
    header_offset: 0,
);

Would work for CSV but not for XML adapter, since internally we are using there XMLReader::open that only accepts string argument not resource (I believe XMLReader is initializing stream under the hood) .

Using file_name to pass remote streams.

(new Flow())
    ->extract(CSV::from_file(
        file_name: 's3://my-bucket-name/test.csv',
        header_offset: 0,
    ))
    // ...
    ->run();

This one would be a bit more flexible but as you noticed we would need to hack a bit file existence validation.
It's also a bit unclear what the argument is because it's called file_name but accepts also stream paths?

I came up with another solution for that problem, so basically etl would need to define a very generic Stream class.

final class Stream
{
     public function __construct(private readonly $path, Mode $mode, private readonly array $options = []);

     public static function local(string $file_path, Mode $mode, array $options) : self 
     {
          // validation
          return new self("file://" . $file_path, $mode, $options); 
     }

     public static function remote(string $path, Mode $mode, array $options) : self 
     {
          // validation
          return new self($path, Mode::read, $mode, $options); 
     }

     public function options() : array;

     public function path() : array;

     public function mode() : Mode;
}

We could use it as an argument for all Loaders/Extractors, it would be literally just a DTO that would pass $path, $mode and $options to adapter implementation. Then on the DSL layer, we could simply enforce some unified function names like for example:

CSV::from(Stream::read_local("/path/to/file.csv"), ...);
CSV::from(Stream::read_remote("abfs://path/to/file.csv"), ...);

CSV::to(Stream::write_local("/path/to/file.csv"), ...);
CSV::to(Stream::write_remote("abfs://path/to/file.csv"), ...);

That would be obviously a BC Break but it would also open all adapters for steams and which brings us to the second idea.

Seekable Remote Streams

We could try to create a custom stream for each adapter provided by Flysystem.
The only problematic part is how to pass options into them, but that's also why I added this last options parameter to Stream.
Internally adapters could just pass options into stream_context_create (in case of XML into libxml_set_streams_context).

So at the end of the day in your case, the CSV DSL would look something like this:

CSV::from(Stream::read_remote("s3://path/to/file.csv", $awsClientOptions), ...);

Under the hood, DSL would pass the Stream instance to selected Adapter.
Adapter depending on the internal implementation would turn Stream into a resource.
StreamWrapper would neet to get registered and then fopen would pretty much initialize and operate on Flysystem adapter allowing any of the data formats to be read/write locally and remotely.

I need to validate across all adapters if libraries used by them can work with resource $stream instead of string $file, list of adapters below.

• CSV
• XML
• Json
• Parquet
• Avro

I just got stuck at writing to single/multiple files. It would need to be something like:

CSV::to_file(Stream::write_remote("abfs://path/to/file.csv"), ...);
CSV::to_directory(Stream::write_remote("abfs://path/to/file.csv"), ...);

and I'm not sure if I like it, CSV::to(Stream::write_remote("abfs://path/to/file.csv"), ...); is cleaner but then in order to stay safe in parallel pipelines we would need to always write to_directory instead of single file.

[norberttech]

So for example Avro adapter is doing fopen internally.

https://github.com/flix-tech/avro-php/blob/master/lib/avro/io.php

I need to check if there is a way to set options without passing context to open but it might be impossible 🙁

[shadowhand]

Couple of things:

1. I don't think every extractor/loader needs to support streams, but the ones that can should have methods like CSV::fromStream($resource) and CSV::toStream($resource). I don't really see how "directory" methods would come into play with streams, since directories cannot be opened as resources. Which is to say, a working solution is better than a future "perfect" solution with no clear path to reach said solution.
2. (Unrelated) I'm really confused about method and parameter naming. It seems like there is a mix of snake_case and studlyCaps but no pattern for each. Before a release, I think it would be good to use PSR-2 (or 12) formatting everywhere for consistency, changing snake to studly.

[norberttech]

I don't think every extractor/loader needs to support streams, but the ones that can should have methods like CSV::fromStream($resource) and CSV::toStream($resource). I don't really see how "directory" methods would come into play with streams, since directories cannot be opened as resources. Which is to say, a working solution is better than a future "perfect" solution with no clear path to reach said solution.

Well, loaders are pretty problematic, lets say you would like to process your data in parallel (multiple processes) then writing to a single stream creates a concurrency issue. Because how 4 subprocesses are supposed to write to a single resource ?
We already support parallel processing through ReactPHP adapter or AMP adapter

Reading recursively from a directory on the other hand is just convenient. I also don't see why reading from multiple streams in a single pipeline would be a problem? We would need to scan the remote filesystem, find all files and put each of them into a standalone extractor that would be combined in a chain extractor.
Your transformers would receive data in chunks anyway and if those chunks would come from one or multiple resources is not important.

(Unrelated) I'm really confused about method and parameter naming. It seems like there is a mix of snake_case and studlyCaps but no pattern for each. Before a release, I think it would be good to use PSR-2 (or 12) formatting everywhere for consistency, changing snake to studly.

Snake case is used on the DSL layer (Domain Specific Language) and the reason for this is readability (it's a subjective reasoning)
Let me just show what I mean by DSL on example:

DSL:

<?php 

(new Flow())
    ->read(CSV::from_file($filePath, 10_000, 0))
    ->rows(Transform::array_unpack('row'))
    ->drop('row')
    ->rows(Transform::to_integer('id'))
    ->rows(Transform::string_concat(['name', 'last name'], ' ', 'name'))
    ->drop('last name')
    ->write(To::output())
    ->limit(10)
    ->run();

Without DSL:

<?php

(new Flow())
    ->read(new CSVExtractor($file_name, 10_000, 0))
    ->rows(new Transformer\ArrayUnpackTransformer('row'))
    ->rows(new RemoveEntriesTransformer('row'))
    ->rows(new CastTransformer(CastToInteger::nullable('id')))
    ->rows(new Transformer\StringConcatTransformer(['name', 'last name'], ' ', 'name'))
    ->rows(new RemoveEntriesTransformer('last name'))
    ->write(Loader\StreamLoader::output())
    ->limit(10)
    ->run()

Long story short, everything in Flow\ETL\DSL is a domain-specific language that follows snake case convention (like PHP), everything else is PSR-2

Obviously using snake_case at DSL was a very subjective decision, I guess camelCase would also work here, or maybe even both.

<?php 

(new Flow())
    ->read(CSV::fromFile($filePath, 10_000, 0))
    ->rows(Transform::arrayUnpack('row'))
    ->drop('row')
    ->rows(Transform::toInteger('id'))
    ->rows(Transform::stringConcat(['name', 'last name'], ' ', 'name'))
    ->drop('last name')
    ->write(To::output())
    ->limit(10)
    ->run();

[shadowhand]

It seems to me that the parallel/async features have created a lot of complexity throughout the app.

Personally, I do not think the DSL "readability" is a strong enough reason to use a different method naming convention. It feels (and looks) awkward with surrounding code.

[norberttech]

It seems to me that the parallel/async features have created a lot of complexity throughout the app.

Indeed, but it's also a very important feature that makes moving data between sources/sinks a lot faster however I think I made some progress on this topic.

League CSV is a pretty complex library, that except reading/writing CSV files is allowing also multiple other operations like doing SQL queries, converting into JSON and XML, and a few other things. That complexity comes with a need to seek over files which makes remote reading/writing pretty impossible. This approach makes sense but not for ETL use case where we just need to read row by row without moving back or jumping to the beginning of the file. That's why I think we gonna replace LeagueCSV with a simpler, custom implementation that will work only on streams.

For now, I did a few tests with Json and S3, here is a code example allowing me to read only 10 lines from 146Mb remote file using Flysystem library together with a custom Steam Wrapper.

<?php

use Aws\S3\S3Client;
use Flow\ETL\Adapter\JSON\JSONMachine\JsonExtractor as ResourceJsonExtractor;
use Flow\ETL\DSL\To;
use Flow\ETL\DSL\Transform;
use Flow\ETL\Flow;
use League\Flysystem\AwsS3V3\AwsS3V3Adapter;
use League\Flysystem\Filesystem;

set_error_handler(function($severity, $message, $file, $line) {
    if (error_reporting() & $severity) {
        throw new \ErrorException($message, 0, $severity, $file, $line);
    }
});

require __DIR__ . '/vendor/autoload.php';

\stream_wrapper_register("flow-s3", "FlowStream");

final class FlowStream
{
    private ?League\Flysystem\Filesystem $fs = null;

    private $resource = null;

    function stream_open($path, $mode, $options, &$opened_path)
    {
        $url = \parse_url($path);

        $options = stream_context_get_options($this->context);

        $client = new S3Client($options['flow-s3']['client']);

        $adapter = new AwsS3V3Adapter($client, $url['host']);
        $this->fs = new Filesystem($adapter);

        $this->resource = $this->fs->readStream($url['path']);

        return true;
    }

    function stream_read($count)
    {
        return \fread($this->resource, $count);
    }

    function stream_write($data)
    {
    }

    function stream_tell()
    {
        return (int) \ftell($this->resource);
    }

    function stream_eof()
    {
        return feof($this->resource);
    }

    function stream_seek($offset, $whence)
    {
        return \fseek($this->resource, $offset, $whence);
    }

    function stream_metadata($path, $option, $var)
    {
        return false;
    }

    function stream_stat()
    {
        return \fstat($this->resource);
    }
}

$res = \fopen('flow-s3://flow-php/dataset.json', 'r', false, stream_context_create($options = [
    'flow-s3' => [
        'seekable' => true,
        'client' => [
            'credentials' => [
                'key'    => '__key__',
                'secret' => '__secret__'
            ],
            'region' => 'eu-west-2',
            'version' => 'latest',
        ]
    ]
]));

(new Flow())
    ->read(new ResourceJsonExtractor($res, 10))
    ->rows(Transform::array_unpack('row'))
    ->drop('row')
    ->rows(Transform::to_integer('id'))
    ->rows(Transform::string_concat(['name', 'last name'], ' ', 'name'))
    ->drop('last name')
    ->write(To::output(false))
    ->limit(10)
    ->run();

Works like a charm, later I will do some more testing with writing JSON files but it seems it should work out of the box.

It seems that Flysystem adapter + generic Stream interface should do the job, and to support parallel writes we would treat external resource path always as a folder in which flow would write files with generated names (file name generator strategy could be an interface).

I'm going to keep working on this, it should let us read/write files from and into all remote filesystems supported by Flysystem

[norberttech]

I'm getting somewhere with this approach, here is some high-level, working proof of concept.

It's reading 10 elements from 146Mb JSON and it's writing them to a new file at a remote filesystem.

[shadowhand]

Great progress!

[norberttech]

I'm happy to announce that FileStream abstraction was added to the core ETL library.
This allowed me to work on Streams Adapter that comes with remote files support which eventually resulted in Json Adapter adjustments allowing to read/write into local/remote files.
An example of reading JSON from AW S3, processing, and then uploading to Azure Blob is available in examples repo.

It also seems that we finally have a unified and pretty straightforward DSL for reading/writing, just take a look at JSON and Stream DSL's.

Just two methods:

JSON::from(FileStream|array<FileStream>|string)
JSON::to(FileStream|array<FileStream>|string)

which works great with:

Stream::local_file() : LocalFile
Stream::local_directory() : array<LocalFile>
Stream::aws_s3_file() : RemoteFile
Stream::aws_s3_directory() : array<RemoteFile>
...

Now we just need to adjust the remaining file format adapters to follow this convention ✌️

[norberttech]

Obviously, it would not be possible without the amazing Flysystem library!
Flysystem it's a giant that carries us on his shoulders ❤️

[norberttech]

And the CSV Adapter was updated (will merge it later today):
flow-php/etl-adapter-csv#108