How can I create an empty dataset from on a PySpark schema in Palantir Foundry?

user_5352227 0
Viewed 78

I have a PySpark schema that describes columns and their types for a dataset (which I could write by hand, or get from an existing dataset by going to the 'Columns' tab, then 'Copy PySpark schema').

I want an empty dataset with this schema, for example that could be used as a backing dataset for a writeback-only ontology object. How can I create this in Foundry?

foundry-code-repositories palantir-foundry
2 Answers

To do this in Python, you can create an empty dataset by using the Spark session from the context to create a DataFrame with the schema, for example:

from pyspark.sql import types as T
from transforms.api import transform_df, configure, Output

SCHEMA = T.StructType([
    T.StructField('entity_name', T.StringType()),
    T.StructField('thing_value', T.IntegerType()),
    T.StructField('created_at', T.TimestampType()),
])


# Given there is no work to do, save on compute by running it on the driver
@configure(profile=["KUBERNETES_NO_EXECUTORS_SMALL"])
@transform_df(
    Output("/some/dataset/path/or/rid"),
)
def compute(ctx):
    return ctx.spark_session.createDataFrame([], schema=SCHEMA)
user_53522270

To do this in Java, you can create a transform using the Spark session on the TransformContext:

package myproject.datasets;

import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.Output;
import com.palantir.transforms.lang.java.api.TransformProfiles;
import com.palantir.transforms.lang.java.api.TransformContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.*;
import java.util.List;

public final class MyTransform {

    private static final StructType SCHEMA = new StructType()
        .add(new StructField("entity_name", DataTypes.StringType, true, Metadata.empty()))
        .add(new StructField("thing_value", DataTypes.IntegerType, true, Metadata.empty()))
        .add(new StructField("created_at", DataTypes.TimestampType, true, Metadata.empty()));

    @Compute
    // Given there is no work to do, save on compute by running it on the driver
    @TransformProfiles({ "KUBERNETES_NO_EXECUTORS_SMALL" })
    @Output("/some/dataset/path/or/rid")
    public Dataset<Row> myComputeFunction(TransformContext context) {
        return context.sparkSession().createDataFrame(List.of(), SCHEMA);
    }
}
user_53522270
Related