AWS Glue pyspark UDF

Posted

技术标签:

【中文标题】AWS Glue pyspark UDF【英文标题】: 【发布时间】:2018-03-08 18:17:15 【问题描述】:

在 AWS Glue 中,我需要将浮点值(摄氏度到华氏度)转换为 UDF。

以下是我的 UDF:

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

我在 spark 数据框中使用 UDF,如下所示:

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

当我运行代码时,我收到错误消息:

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 的新手,并且我的新列架构没有创建,并且是空的。

用于上述示例的代码片段是:

%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): "

谢谢

【问题讨论】:

你用的是什么版本的python? 它的 Python2.7,使用 AWS Glue 的开发端点 您能否尝试提供足够的资源来创建minimal reproducible example?例如,not_found 是什么? How to make good reproducible apache spark dataframe examples. 另外,您不需要删除该列。您可以使用withColumn() 就地修改列。有关最近的问题,请参阅 my answer。 谢谢 pault,我已经用代码 sn-p 更新了我的问题,我使用的数据帧的架构详细信息,供您参考。认为我在 UDF 及其用法上出错了。请建议正确的 UDF 和用法。 【参考方案1】:

上面的解决方案(摄氏到华氏),以备参考:

#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应用于华氏度后:

+-----------+--------------------+--------+---------+---------+----------+-----+----+----+----+----+
|    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|
+-----------+--------------------+--------+---------+---------+----------+-----+----+----+----+----+

【讨论】:

以上是关于AWS Glue pyspark UDF的主要内容,如果未能解决你的问题,请参考以下文章

AWS Glue pyspark UDF

从 AWS Glue/PySpark 中的 100 个表中选择数据

如何使用带有 PySpark 的 WHERE 子句在 AWS Glue 中查询 JDBC 数据库?

AWS Glue PySpark 无法计算记录

如何在 AWS Glue pyspark 脚本中合并两个节点

Python/Pyspark 迭代代码(用于 AWS Glue ETL 作业)