Data Sources and Extensions

from pyspark.sql.datasource import DataSource, DataSourceReader

class CustomDataSource(DataSource):

    def reader(self, schema):

        return CustomReader(schema, self.options)

class CustomReader(DataSourceReader):

    def partitions(self):

        return [None]

    def read(self, partition):

        for row in self._load_records(self.options.get("path")):

            yield row

spark.dataSource.register(CustomDataSource, "custom_source")

df = spark.read.format("custom_source").option("path", "/path/to/data").load()

df.show()

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("XML Connector").getOrCreate()

df = spark.read \

    .format("xml") \

    .option("rowTag", "record") \

    .load("/path/to/data.xml")

df.write \

    .format("xml") \

    .option("rootTag", "records") \

    .option("rowTag", "record") \

    .save("/path/to/output.xml")

from pyspark.sql.datasource import DataSource

class CustomDSV2(DataSource):

    def reader(self, schema):

        return MyDSV2Reader(schema, self.options)

    def writer(self, schema, overwrite):

        return MyDSV2Writer(self.options)

spark.dataSource.register(CustomDSV2, "custom_dsv2")

df = spark.read.format("custom_dsv2").option("option_key", "value").load()

df.write.format("custom_dsv2").option("option_key", "value").save()

CREATE TABLE orders (id INT, amount DOUBLE, status STRING) USING delta;

INSERT INTO orders VALUES (1, 99.99, 'pending'), (2, 149.50, 'completed');

-- Time travel

SELECT * FROM orders VERSION AS OF 0;

SELECT * FROM orders TIMESTAMP AS OF '2025-01-01T00:00:00';

-- Optimize with liquid clustering

ALTER TABLE orders CLUSTER BY (status);