다음 예에서는 동일한 변환을 수행합니다. 하지만 두 번째 예제(SparkSQL)가 가장 깔끔하고 가장 효율적이며, 그 다음은 Pandas UDF이며, 첫 번째 예제의 하위 수준 매핑이 마지막입니다. 다음 예제는 두 열을 합산하는 간단한 변환의 전체 예입니다.
from awsglue import DynamicFrame
# You can have other auxiliary variables, functions or classes on this file, it won't affect the runtime
def record_sum(rec, col1, col2, resultCol):
rec[resultCol] = rec[col1] + rec[col2]
return rec
# The number and name of arguments must match the definition on json config file
# (expect self which is the current DynamicFrame to transform
# If an argument is optional, you need to define a default value here
# (resultCol in this example is an optional argument)
def custom_add_columns(self, col1, col2, resultCol="result"):
# The mapping will alter the columns order, which could be important
fields = [field.name for field in self.schema()]
if resultCol not in fields:
# If it's a new column put it at the end
fields.append(resultCol)
return self.map(lambda record: record_sum(record, col1, col2, resultCol)).select_fields(paths=fields)
# The name we assign on DynamicFrame must match the configured "functionName"
DynamicFrame.custom_add_columns = custom_add_columns
다음 예제는 SparkSQL API를 활용하는 동일한 변환입니다.
from awsglue import DynamicFrame
# The number and name of arguments must match the definition on json config file
# (expect self which is the current DynamicFrame to transform
# If an argument is optional, you need to define a default value here
# (resultCol in this example is an optional argument)
def custom_add_columns(self, col1, col2, resultCol="result"):
df = self.toDF()
return DynamicFrame.fromDF(
df.withColumn(resultCol, df[col1] + df[col2]) # This is the conversion logic
, self.glue_ctx, self.name)
# The name we assign on DynamicFrame must match the configured "functionName"
DynamicFrame.custom_add_columns = custom_add_columns
다음 예제에서는 동일한 변환을 사용하지만 pandas UDF를 사용하므로 일반 UDF를 사용하는 것보다 더 효율적입니다. pandas UDF 작성에 대한 자세한 내용은 Apache Spark SQL 설명서
from awsglue import DynamicFrame
import pandas as pd
from pyspark.sql.functions import pandas_udf
# The number and name of arguments must match the definition on json config file
# (expect self which is the current DynamicFrame to transform
# If an argument is optional, you need to define a default value here
# (resultCol in this example is an optional argument)
def custom_add_columns(self, col1, col2, resultCol="result"):
@pandas_udf("integer") # We need to declare the type of the result column
def add_columns(value1: pd.Series, value2: pd.Series) → pd.Series:
return value1 + value2
df = self.toDF()
return DynamicFrame.fromDF(
df.withColumn(resultCol, add_columns(col1, col2)) # This is the conversion logic
, self.glue_ctx, self.name)
# The name we assign on DynamicFrame must match the configured "functionName"
DynamicFrame.custom_add_columns = custom_add_columns