AWS Glue pyspark UDF

本文介绍了AWS Glue pyspark UDF的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

在AWS Glue中，我需要转换一个浮点值(摄氏度到华氏度)，并且正在使用UDF.

In AWS Glue, I need to convert a float value (celsius to fahrenheit) and am using an UDF.

以下是我的UDF:

toFahrenheit = udf(lambda x: '-1' if x in not_found else x * 9 / 5 + 32, StringType())

我在spark数据框中使用UDF的方式如下:

I am using the UDF as follows, in the spark dataframe:

weather_df.withColumn("new_tmax", toFahrenheit(weather_df["tmax"])).drop("tmax").withColumnRenamed("new_tmax","tmax")

运行代码时，我收到的错误消息为:

When I run the code, am getting the error message as :

IllegalArgumentException: u"requirement failed: The number of columns doesn't match.\nOld column names (11): station, name, latitude, longitude, elevation, date, awnd, prcp, snow, tmin, tmax\nNew column names (0): "

不确定如何调用UDF(这是python/pyspark的新功能)，并且未创建新的列架构，并且为空.

Not sure how to invoke the UDF, as am new to python / pyspark, and my new column schema is not created, and empty.

上面示例中的代码片段为:

The code snipped used for above sample is :

%pyspark import sys from pyspark.context import SparkContext from awsglue.context import GlueContext from awsglue.context import DynamicFrame from awsglue.transforms import * from awsglue.utils import getResolvedOptions from awsglue.job import Job from pyspark.sql import SparkSession from pyspark.sql.functions import udf from pyspark.sql.types import StringType glueContext = GlueContext(SparkContext.getOrCreate()) weather_raw = glueContext.create_dynamic_frame.from_catalog(database = "ohare-airport-2006", table_name = "ohare_intl_airport_2006_08_climate_csv") print "cpnt : ", weather_raw.count() weather_raw.printSchema() weather_raw.toDF().show(10) #UDF to convert the air temperature from celsius to fahrenheit (For sample transformation) #toFahrenheit = udf((lambda c: c[1:], c * 9 / 5 + 32) toFahrenheit = udf(lambda x: '-1' if x in not_found_cat else x * 9 / 5 + 32, StringType()) #Apply the UDF to maximum and minimum air temperature wthdf = weather_df.withColumn("new_tmin", toFahrenheit(weather_df["tmin"])).withColumn("new_tmax", toFahrenheit(weather_df["tmax"])).drop("tmax").drop("tmin").withColumnRenamed("new_tmax","tmax").withColumnRenamed("new_tmin","tmin") wthdf.toDF().show(5)

模式

weather_df: root |-- station: string |-- name: string |-- latitude: double |-- longitude: double |-- elevation: double |-- date: string |-- awnd: double |-- fmtm: string |-- pgtm: string |-- prcp: double |-- snow: double |-- snwd: long |-- tavg: string |-- tmax: long |-- tmin: long

错误跟踪:

Traceback (most recent call last): File "/tmp/zeppelin_pyspark-3684249459612979499.py", line 349, in <module> raise Exception(traceback.format_exc()) Exception: Traceback (most recent call last): File "/tmp/zeppelin_pyspark-3684249459612979499.py", line 342, in <module> exec(code) File "<stdin>", line 3, in <module> File "/usr/lib/spark/python/pyspark/sql/dataframe.py", line 1558, in toDF jdf = self._jdf.toDF(self._jseq(cols)) File "/usr/lib/spark/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py", line 1133, in __call__ answer, self.gateway_client, self.target_id, self.name) File "/usr/lib/spark/python/pyspark/sql/utils.py", line 79, in deco raise IllegalArgumentException(s.split(': ', 1)[1], stackTrace) IllegalArgumentException: u"requirement failed: The number of columns doesn't match.\nOld column names (11): station, name, latitude, longitude, elevation, date, awnd, prcp, snow, tmin, tmax\nNew column names (0): "

谢谢

推荐答案

上述解决方案(从Celcius到Fahrenheit)，以防万一供参考:

Solution for the above (Celcius to Fahrenheit), just in case for reference:

#UDF to convert the air temperature from celsius to fahrenheit toFahrenheit = udf(lambda x: x * 9 / 5 + 32, StringType()) weather_in_Fahrenheit = weather_df.withColumn("new_tmax", toFahrenheit(weather_df["tmax"])).withColumn("new_tmin", toFahrenheit(weather_df["tmin"])).drop("tmax").drop("tmin").withColumnRenamed("new_tmax","tmax").withColumnRenamed("new_tmin","tmin") weather_in_Fahrenheit.show(5)

原始数据示例:

+-----------+--------------------+---------+--------+---------+----+----+----+----+----------+ | station| name|elevation|latitude|longitude|prcp|snow|tmax|tmin| date| +-----------+--------------------+---------+--------+---------+----+----+----+----+----------+ |USW00094846|CHICAGO OHARE INT...| 201.8| 41.995| -87.9336| 0.0| 0.0| 25| 11|2013-01-01| |USW00094846|CHICAGO OHARE INT...| 201.8| 41.995| -87.9336| 0.0| 0.0| 30| 10|2013-01-02| |USW00094846|CHICAGO OHARE INT...| 201.8| 41.995| -87.9336| 0.0| 0.0| 29| 18|2013-01-03| |USW00094846|CHICAGO OHARE INT...| 201.8| 41.995| -87.9336| 0.0| 0.0| 36| 13|2013-01-04| |USW00094846|CHICAGO OHARE INT...| 201.8| 41.995| -87.9336|0.03| 0.4| 39| 18|2013-01-05| |USW00094846|CHICAGO OHARE INT...| 201.8| 41.995| -87.9336| 0.0| 0.0| 36| 18|2013-01-06| |USW00094846|CHICAGO OHARE INT...| 201.8| 41.995| -87.9336| 0.0| 0.0| 41| 15|2013-01-07| |USW00094846|CHICAGO OHARE INT...| 201.8| 41.995| -87.9336| 0.0| 0.0| 44| 22|2013-01-08| |USW00094846|CHICAGO OHARE INT...| 201.8| 41.995| -87.9336| 0.0| 0.0| 50| 27|2013-01-09| |USW00094846|CHICAGO OHARE INT...| 201.8| 41.995| -87.9336|0.63| 0.0| 45| 22|2013-01-10| +-----------+--------------------+---------+--------+---------+----+----+----+----+----------+

将UDF应用到华氏度之后:

After applying the UDF toFahrenheit:

+-----------+--------------------+--------+---------+---------+----------+-----+----+----+----+----+ | station| name|latitude|longitude|elevation| date| awnd|prcp|snow|tmax|tmin| +-----------+--------------------+--------+---------+---------+----------+-----+----+----+----+----+ |USW00094846|CHICAGO OHARE INT...| 41.995| -87.9336| 201.8|2013-01-01| 8.5| 0.0| 0.0| 77| 51| |USW00094846|CHICAGO OHARE INT...| 41.995| -87.9336| 201.8|2013-01-02| 8.05| 0.0| 0.0| 86| 50| |USW00094846|CHICAGO OHARE INT...| 41.995| -87.9336| 201.8|2013-01-03|11.41| 0.0| 0.0| 84| 64| |USW00094846|CHICAGO OHARE INT...| 41.995| -87.9336| 201.8|2013-01-04| 13.2| 0.0| 0.0| 96| 55| |USW00094846|CHICAGO OHARE INT...| 41.995| -87.9336| 201.8|2013-01-05| 9.62|0.03| 0.4| 102| 64| +-----------+--------------------+--------+---------+---------+----------+-----+----+----+----+----+