代码示例:使用 ResolveChoice、Lambda 和 ApplyMapping 进行数据准备 - Amazon连接词
AWS 文档中描述的 AWS 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 AWS 服务入门

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

代码示例:使用 ResolveChoice、Lambda 和 ApplyMapping 进行数据准备

本示例中使用的数据集包含医疗保险提供商付款数据,这些数据从两个Data.CMS.gov站点:Instem Provider Provider Provider Provider Summary Provider Summary Provider Summary Provider Summary Pro FY2011) 和2011 年度住院收费数据。下载该数据后,我们修改了它,以在文件末尾引入了几个错误的记录。这个修改过的文件位于上的公有 Amazon S3 存储桶中。s3://awsglue-datasets/examples/medicare/Medicare_Hospital_Provider.csv

源代码在data_cleaning_and_lambda.py中的AmazonGlue 示例GitHub 存储库。

将生成的数据写入单独的 Apache Parquet 文件以供日后分析。将生成的数据写入单独的 Apache Parquet 文件以供日后分析。有关更多信息,请参阅 查看开发终端节点属性

第 1 步:在 Amazon S3 存储桶中爬网数据

  1. 登录到Amazon Web Services Management Console并打开AmazonGlue 控制台https://console.aws.amazon.com/glue/

  2. 按照在 Amazon Glue 控制台上使用爬网程序,创建可爬网s3://awsglue-datasets/examples/medicare/Medicare_Hospital_Provider.csv文件,并且可以将生成的元数据放入名为payments中的AmazonGlue 数据目录。

  3. 运行新爬网程序,然后检查 payments 数据库。在读取该文件的开头以确定其格式和分隔符之后,您应该发现爬网程序已经在数据库中创建了一个名为 medicare 的元数据表。

    medicare 表的架构如下所示:

    Column name Data type ================================================== drg definition string provider id bigint provider name string provider street address string provider city string provider state string provider zip code bigint hospital referral region description string total discharges bigint average covered charges string average total payments string average medicare payments string

第 2 步:将样板脚本添加到开发终结点笔记本中。

将以下样板文件脚本粘贴到开发终端节点笔记本中以导入Amazon您需要的 Glue 库,并设置单个GlueContext

import sys from awsglue.transforms import * from awsglue.utils import getResolvedOptions from pyspark.context import SparkContext from awsglue.context import GlueContext from awsglue.job import Job glueContext = GlueContext(SparkContext.getOrCreate())

第 3 步:比较不同的架构解析

接下来,您可以查看由 Apache Spark 识别的架构是否DataFrame与您的AmazonGlue 履带记录。运行此代码:

medicare = spark.read.format( "com.databricks.spark.csv").option( "header", "true").option( "inferSchema", "true").load( 's3://awsglue-datasets/examples/medicare/Medicare_Hospital_Provider.csv') medicare.printSchema()

下面是来自 printSchema 调用的输出:

root |-- DRG Definition: string (nullable = true) |-- Provider Id: string (nullable = true) |-- Provider Name: string (nullable = true) |-- Provider Street Address: string (nullable = true) |-- Provider City: string (nullable = true) |-- Provider State: string (nullable = true) |-- Provider Zip Code: integer (nullable = true) |-- Hospital Referral Region Description: string (nullable = true) |-- Total Discharges : integer (nullable = true) |-- Average Covered Charges : string (nullable = true) |-- Average Total Payments : string (nullable = true) |-- Average Medicare Payments: string (nullable = true)

接下来,查看一个Amazon连接词DynamicFrame生成:

medicare_dynamicframe = glueContext.create_dynamic_frame.from_catalog( database = "payments", table_name = "medicare") medicare_dynamicframe.printSchema()

printSchema 中的输出如下所示:

root |-- drg definition: string |-- provider id: choice | |-- long | |-- string |-- provider name: string |-- provider street address: string |-- provider city: string |-- provider state: string |-- provider zip code: long |-- hospital referral region description: string |-- total discharges: long |-- average covered charges: string |-- average total payments: string |-- average medicare payments: string

DynamicFrame 生成一个架构,在其中 provider id 可以是 longstring 类型。这些区域有:DataFrame架构列表Provider Id作为一个string类型,并且 “数据目录” 列表provider id作为一个bigint

哪一个是正确的? 文件末尾有两条记录 (共计 16 万条记录),该列中有 string 值。这些是为说明问题而引入的错误记录。

为了解决此类问题,Amazon连接词DynamicFrame介绍了choice。在这种情况下,DynamicFrame 显示 longstring 值都出现在该列中。这些区域有:AmazonGlue 爬虫错过了string值,因为它仅被视为数据的一个 2 MB 前缀。Apache Spark DataFrame 考虑了整个数据集,但它被迫将最一般的类型分配给该列,即 string。事实上,当存在复杂类型或不熟悉的变体时,Spark 通常会采用最一般的情况。

要查询 provider id 列,请先解析选择类型。您可以在 DynamicFrame 中使用 resolveChoice 转换方法,通过 cast:long 选项将这些 string 值转换为 long 值:

medicare_res = medicare_dynamicframe.resolveChoice(specs = [('provider id','cast:long')]) medicare_res.printSchema()

printSchema 输出现在是:

root |-- drg definition: string |-- provider id: long |-- provider name: string |-- provider street address: string |-- provider city: string |-- provider state: string |-- provider zip code: long |-- hospital referral region description: string |-- total discharges: long |-- average covered charges: string |-- average total payments: string |-- average medicare payments: string

其中值是string不能被投下的AmazonGlue 插入null

另一个选项是将选择类型转换为一个 struct,以保持两种类型的值。

接下来,查看异常的行:

medicare_res.toDF().where("'provider id' is NULL").show()

您看到以下内容:

+--------------------+-----------+---------------+-----------------------+-------------+--------------+-----------------+------------------------------------+----------------+-----------------------+----------------------+-------------------------+ | drg definition|provider id| provider name|provider street address|provider city|provider state|provider zip code|hospital referral region description|total discharges|average covered charges|average total payments|average medicare payments| +--------------------+-----------+---------------+-----------------------+-------------+--------------+-----------------+------------------------------------+----------------+-----------------------+----------------------+-------------------------+ |948 - SIGNS & SYM...| null| INC| 1050 DIVISION ST| MAUSTON| WI| 53948| WI - Madison| 12| $11961.41| $4619.00| $3775.33| |948 - SIGNS & SYM...| null| INC- ST JOSEPH| 5000 W CHAMBERS ST| MILWAUKEE| WI| 53210| WI - Milwaukee| 14| $10514.28| $5562.50| $4522.78| +--------------------+-----------+---------------+-----------------------+-------------+--------------+-----------------+------------------------------------+----------------+-----------------------+----------------------+-------------------------+

现在,删除两个格式错误的记录,如下所示:

medicare_dataframe = medicare_res.toDF() medicare_dataframe = medicare_dataframe.where("'provider id' is NOT NULL")

第 4 步:数据映射和使用 Apache Spark Lambda 函数

AmazonGlue 尚未直接支持 Lambda 函数,也称为用户定义函数。但是,您始终可以将 DynamicFrame 和 Apache Spark DataFrame 相互转换,以便除 DynamicFrames 的特殊功能外,还能利用 Spark 功能。

接下来,将付款信息转化为数字,因此 Amazon Redshift 或 Amazon Athena 这样的分析引擎可以更快地进行数字处理:

from pyspark.sql.functions import udf from pyspark.sql.types import StringType chop_f = udf(lambda x: x[1:], StringType()) medicare_dataframe = medicare_dataframe.withColumn( "ACC", chop_f( medicare_dataframe["average covered charges"])).withColumn( "ATP", chop_f( medicare_dataframe["average total payments"])).withColumn( "AMP", chop_f( medicare_dataframe["average medicare payments"])) medicare_dataframe.select(['ACC', 'ATP', 'AMP']).show()

show 调用中的输出如下所示:

+--------+-------+-------+ | ACC| ATP| AMP| +--------+-------+-------+ |32963.07|5777.24|4763.73| |15131.85|5787.57|4976.71| |37560.37|5434.95|4453.79| |13998.28|5417.56|4129.16| |31633.27|5658.33|4851.44| |16920.79|6653.80|5374.14| |11977.13|5834.74|4761.41| |35841.09|8031.12|5858.50| |28523.39|6113.38|5228.40| |75233.38|5541.05|4386.94| |67327.92|5461.57|4493.57| |39607.28|5356.28|4408.20| |22862.23|5374.65|4186.02| |31110.85|5366.23|4376.23| |25411.33|5282.93|4383.73| | 9234.51|5676.55|4509.11| |15895.85|5930.11|3972.85| |19721.16|6192.54|5179.38| |10710.88|4968.00|3898.88| |51343.75|5996.00|4962.45| +--------+-------+-------+ only showing top 20 rows

这些仍然是数据中的字符串。我们可以使用强大的 apply_mapping 转换方法来删除、重命名、转换和嵌套数据,以便其他数据编程语言和系统可以轻松地访问它:

from awsglue.dynamicframe import DynamicFrame medicare_tmp_dyf = DynamicFrame.fromDF(medicare_dataframe, glueContext, "nested") medicare_nest_dyf = medicare_tmp_dyf.apply_mapping([('drg definition', 'string', 'drg', 'string'), ('provider id', 'long', 'provider.id', 'long'), ('provider name', 'string', 'provider.name', 'string'), ('provider city', 'string', 'provider.city', 'string'), ('provider state', 'string', 'provider.state', 'string'), ('provider zip code', 'long', 'provider.zip', 'long'), ('hospital referral region description', 'string','rr', 'string'), ('ACC', 'string', 'charges.covered', 'double'), ('ATP', 'string', 'charges.total_pay', 'double'), ('AMP', 'string', 'charges.medicare_pay', 'double')]) medicare_nest_dyf.printSchema()

printSchema 输出如下所示:

root |-- drg: string |-- provider: struct | |-- id: long | |-- name: string | |-- city: string | |-- state: string | |-- zip: long |-- rr: string |-- charges: struct | |-- covered: double | |-- total_pay: double | |-- medicare_pay: double

将数据重新变成 Spark DataFrame 后,您可以显示它现在的外观:

medicare_nest_dyf.toDF().show()

您可以在一个 (扩展) 代码行中执行所有这些操作:

+--------------------+--------------------+---------------+--------------------+ | drg| provider| rr| charges| +--------------------+--------------------+---------------+--------------------+ |039 - EXTRACRANIA...|[10001,SOUTHEAST ...| AL - Dothan|[32963.07,5777.24...| |039 - EXTRACRANIA...|[10005,MARSHALL M...|AL - Birmingham|[15131.85,5787.57...| |039 - EXTRACRANIA...|[10006,ELIZA COFF...|AL - Birmingham|[37560.37,5434.95...| |039 - EXTRACRANIA...|[10011,ST VINCENT...|AL - Birmingham|[13998.28,5417.56...| |039 - EXTRACRANIA...|[10016,SHELBY BAP...|AL - Birmingham|[31633.27,5658.33...| |039 - EXTRACRANIA...|[10023,BAPTIST ME...|AL - Montgomery|[16920.79,6653.8,...| |039 - EXTRACRANIA...|[10029,EAST ALABA...|AL - Birmingham|[11977.13,5834.74...| |039 - EXTRACRANIA...|[10033,UNIVERSITY...|AL - Birmingham|[35841.09,8031.12...| |039 - EXTRACRANIA...|[10039,HUNTSVILLE...|AL - Huntsville|[28523.39,6113.38...| |039 - EXTRACRANIA...|[10040,GADSDEN RE...|AL - Birmingham|[75233.38,5541.05...| |039 - EXTRACRANIA...|[10046,RIVERVIEW ...|AL - Birmingham|[67327.92,5461.57...| |039 - EXTRACRANIA...|[10055,FLOWERS HO...| AL - Dothan|[39607.28,5356.28...| |039 - EXTRACRANIA...|[10056,ST VINCENT...|AL - Birmingham|[22862.23,5374.65...| |039 - EXTRACRANIA...|[10078,NORTHEAST ...|AL - Birmingham|[31110.85,5366.23...| |039 - EXTRACRANIA...|[10083,SOUTH BALD...| AL - Mobile|[25411.33,5282.93...| |039 - EXTRACRANIA...|[10085,DECATUR GE...|AL - Huntsville|[9234.51,5676.55,...| |039 - EXTRACRANIA...|[10090,PROVIDENCE...| AL - Mobile|[15895.85,5930.11...| |039 - EXTRACRANIA...|[10092,D C H REGI...|AL - Tuscaloosa|[19721.16,6192.54...| |039 - EXTRACRANIA...|[10100,THOMAS HOS...| AL - Mobile|[10710.88,4968.0,...| |039 - EXTRACRANIA...|[10103,BAPTIST ME...|AL - Birmingham|[51343.75,5996.0,...| +--------------------+--------------------+---------------+--------------------+ only showing top 20 rows

第 5 步:将数据写入到 Apache 实木复合地板

AmazonGlue 可以很容易地以诸如 Apache Parquet 这样的格式编写数据,以便关系数据库可以有效地使用它:

glueContext.write_dynamic_frame.from_options( frame = medicare_nest_dyf, connection_type = "s3", connection_options = {"path": "s3://glue-sample-target/output-dir/medicare_parquet"}, format = "parquet")