Add pyspark dataframe_loader

[DavidKatz-il]

Add dataframe_loader to support inputs.

[natekupp]

hey @sryza can you take a look at this one too?

[sryza]

Thanks for your contribution @DavidKatz-il ! I'm impressed with your thoroughness.

An important area to think through is that Dagster should be able to support any version of Spark. I'm concerned that, by wrapping every individual option to every Spark input format in Dagster config, it will tie us too closely to specific Spark versions. I.e. Spark might add or remove an option, and then it becomes unclear which Spark version Dagster's configuration should reflect.

Although it doesn't give us the same advantage of having all the options nicely display in the Dagit config editor, I think a safer approach would be to use permissive dictionaries for the set of options within each input format. E.g.

'jdbc': {},
'orc': {},

This allows Spark itself to remain as the arbiter of what config options are available.

[DavidKatz-il]

Hey @sryza, thanks for reviewing.
Right now the user can add options that we didn't configure because we are using Permissive.
So you suggesting to put an empty Permissive?
e.g.

'jdbc': Permissive(),
'orc': Permissive(),
...

If so, what about options that are required, e.g. 'path'.

[sryza]

Thought about it a little bit more, and I think you're right. As long as we use Permissive, it's helpful to include the detailed config.

I just launched a build for the PR.

[flvndh]

Great contribution @DavidKatz-il! What do you think of adding support for Apache Iceberg and Delta Lake as well ?

[DavidKatz-il]

Hi @flvndh, Does it require packages outside pyspark?

[flvndh]

@DavidKatz-il just jars packages : org.apache.iceberg:iceberg-spark3-runtime:0.9.0 and io.delta:delta-core_2.12:0.7.0.

[DavidKatz-il]

@sryza what do you think?

[sryza]

This change looks ready to go in. I think adding support for Iceberg and Delta Lake would be cool. @DavidKatz-il - if you're interested in adding those to this PR, then great. If you'd like me to merge this as-is, also great.

[DavidKatz-il]

I added the file_type 'other'.

e.g.

'other': {    
    'path': 'path_to_file',    
    'format': 'delta',    
    ...
}

[sryza]

would you mind adding tests for the "other" cases for input and output?

[sryza]

@schrockn brought up the question of whether we should avoid the "other" escape hatch in favor of making it easy to compose type materializers: https://dagster.phacility.com/D4188.

@DavidKatz-il - it could take some time for us to reach some resolution on that, but I still think the core of this PR, which adds support for the native Spark formats, is useful. If you want to take out the "other" option for now, I can merge the rest of it. Thanks for bearing with us on the thrash here.

EDIT: it looks like we actually got to a resolution on that issue already ^^, and the escape hatch you added here seems like a reasonable direction. It would still be good to have tests for it.

[sryza]

I'm noticing a syntax error here for in the python 2.7 tests: https://buildkite.com/dagster/dagster/builds/14227#9a8db1d9-518e-4491-87a8-d247ca90946c

[DavidKatz-il]

The comma at the end of the line caused this error on python 2.7.
 

[sryza]

This looks good. Merging. Thanks for your contribution @DavidKatz-il !