I am moving SAS code to pyspark. Amongst miscellaneous data transformation, there are spline derivations like this:
var_s1 = max(0, min(1, (var-440)/( 600-440))) - max(0, min(1, (680-440)/( 600-440)));
var_s2 = max(0, min(1, (var-600)/( 800-600))) - max(0, min(1, (680-600)/( 800-600)));
var_s3 = max(0, min(1, (var-800)/(1000-800))) - max(0, min(1, (680-800)/(1000-800)));
Then I try to use an UDF to do the splines:
@udf(returnType=DoubleType())
def spline_cf(invar, c1, c2, c3, c4, c5, c6, c7):
return greatest(0, least(1, (invar - c1) / (c2 - c3))) - greatest(0, least(1, (c4-c5) / (c6-c7)))
data = (
data
.withColumn('var_s1', spline_cf(col('var'), lit(440), lit( 600), lit(440), lit(680), lit(440), lit( 600), lit(440)))
.withColumn('var_s2', spline_cf(col('var'), lit(600), lit( 800), lit(600), lit(680), lit(600), lit( 800), lit(600)))
.withColumn('var_s3', spline_cf(col('var'), lit(800), lit(1000), lit(800), lit(680), lit(800), lit(1000), lit(800)))
)
The end result of this attempt is:.
...
return Column(sc._jvm.functions.least(_to_seq(sc, cols, _to_java_column)))
AttributeError: 'NoneType' object has no attribute '_jvm'
What could I be doing wrong?