构建数据湖的一款开源利器 – AWS Data Wrangler

1.	import awswrangler as wr  
2.	import pandas as pd  
3.	  
4.	# 定义DataFrame  
5.	df = pd.DataFrame({"id": [1, 2], "value": ["foo", "boo"]})  
6.	  
7.	# 在S3的数据湖上存储数据  
8.	wr.s3.to_parquet(  
9.	    df=df,  
10.	    path="s3://bucket/dataset/",  
11.	    dataset=True,  
12.	    database="my_db",  
13.	    table="my_table"  
14.	)  
15.	  
16.	# 直接从 Amazon S3 中检索数据  
17.	df = wr.s3.read_parquet("s3://bucket/dataset/", dataset=True)  
18.	  
19.	# #从 Amazon Athena 中检索数据  
20.	df = wr.athena.read_sql_query("SELECT * FROM my_table", database="my_db")  
21.	  
22.	# 从Glue进行Redshift连接（SQLAlchemy），并从Redshift Spectrum检索数据  
23.	engine = wr.catalog.get_engine("my-redshift-connection")  
24.	df = wr.db.read_sql_query("SELECT * FROM external_schema.my_table", con=engine)  
25.	  
26.	# 创建QuickSight数据源和数据集来映射新表  
27.	wr.quicksight.create_athena_data_source("athena-source", allowed_to_manage=["username"])  
28.	wr.quicksight.create_athena_dataset(  
29.	    name="my-dataset",  
30.	    database="my_db",  
31.	    table="my_table",  
32.	    data_source_name="athena-source",  
33.	    allowed_to_manage=["username"]  
34.	)  
35.	  
36.	# 从Glue目录获取MySQL连接(SQLAlchemy)并将数据加载到MySQL中  
37.	engine = wr.catalog.get_engine("my-mysql-connection")  
38.	wr.db.to_sql(df, engine, schema="test", name="my_table")  
39.	  
40.	# 从Glue目录获取PostgreSQL连接(SQLAlchemy)并将数据加载到PostgreSQL  
41.	engine = wr.catalog.get_engine("my-postgresql-connection")  
42.	wr.db.to_sql(df, engine, schema="test", name="my_table")  

1.	import os  
2.	import json  
3.	import boto3  
4.	import logging  
5.	import urllib.parse  
6.	import awswrangler as wr  
7.	import pandas as pd  
8.	from datetime import datetime  
9.	from botocore.exceptions import ClientError  
10.	  
11.	  
12.	logger = logging.getLogger(__name__)  
13.	  
14.	  
15.	def s3_url(bucket, prefix, folder=False):  
16.	    """ Produce s3 url from 'bucket' and 'prefix' 
17.	    """  
18.	    assert bucket and prefix  
19.	  
20.	    return "s3://{bucket}/{prefix}{folder}".format(  
21.	        bucket=bucket,  
22.	        prefix=prefix,  
23.	        folder="/" if folder else "")  
24.	  
25.	  
26.	def read_parquet(s3_path, columns, source, dataset=True):  
27.	    """ Read Apache Parquet file(s) from a received S3  
28.	        prefix or list of S3 objects paths. 
29.	        Convert the last_updated column from Timestamp to Epoch time. 
30.	        Return records as list of dictionary.  
31.	    """  
32.	    assert source, "source can't be None"  
33.	  
34.	    df = wr.s3.read_parquet(path=s3_path, columns=columns, dataset=dataset)  
35.	    df[["last_updated"]] = (df[["last_updated"]] - pd.Timestamp("1970-01-01")) // pd.Timedelta('1s')  
36.	    df["source"] = source  
37.	    df["location"] = s3_path  
38.	      
39.	    return df.to_dict(orient='records')  
40.	  
41.	  
42.	def write_to_ddb(client, table_name, records):  
43.	    """ Write data to DynamoDB table. 
44.	        Returns inserted row count. 
45.	    """  
46.	  
47.	    table = client.Table(table_name)  
48.	    row_count = 0  
49.	    for row in records:  
50.	        table.put_item(Item=row)  
51.	        row_count += 1  
52.	  
53.	    return row_count  
54.	  
55.	  
56.	def lambda_handler(event, context):  
57.	  
58.	    record = event["Records"][0]  
59.	  
60.	    assert "eventSource" in record  
61.	    assert "eventName" in record  
62.	    assert record["eventSource"] == "aws:s3", "Unknown event source: %s." % record["eventSource"]  
63.	    assert record["eventName"].startswith(  
64.	        "ObjectCreated:"), "Unsupported event name: %s." % record["eventName"]  
65.	    assert "s3" in record  
66.	    assert "bucket" in record["s3"]  
67.	    assert "name" in record["s3"]["bucket"]  
68.	    assert "object" in record["s3"]  
69.	    assert "key" in record["s3"]["object"]  
70.	  
71.	    bucket = record["s3"]["bucket"]["name"]  
72.	  
73.	    # If S3 URL contains special character usch as '=', it will be quoted, like: %3D  
74.	    # This is to unquote them back to original character, or it will complain path not exist.  
75.	    key = urllib.parse.unquote(record["s3"]["object"]["key"])  
76.	  
77.	    #create s3 path  
78.	    s3_path = s3_url(bucket=bucket, prefix=key)  
79.	  
80.	    # Retrieving the data directly from Amazon S3  
81.	    cols = ["ticker_symbol", "sector", "last_updated"]  
82.	      
83.	    df = read_parquet(s3_path=s3_path,   
84.	                      columns=cols,  
85.	                      source="system",   
86.	                      dataset=True)  
87.	  
88.	    #Instantiate Dynamodb connection  
89.	    ddb = boto3.resource("dynamodb")  
90.	  
91.	    #Get the dynamodb table name  
92.	    ddb_table = os.environ["DDB_TABLE"]  
93.	  
94.	    #Write data to dynamodb  
95.	    record_count = write_to_ddb(  
96.	        client=ddb, table_name=ddb_table, records=df)  
97.	  
98.	    #return total record inserted  
99.	    return("Total records inserted:", record_count)  

1.	#!/bin/bash  
2.	  
3.	set -e  
4.	  
5.	# OVERVIEW  
6.	# This script installs a single pip package in all SageMaker conda environments, apart from the JupyterSystemEnv which  
7.	# is a system environment reserved for Jupyter.  
8.	# Note this may timeout if the package installations in all environments take longer than 5 mins, consider using  
9.	# "nohup" to run this as a background process in that case.  
10.	  
11.	sudo -u ec2-user -i <<'EOF'  
12.	  
13.	# PARAMETERS  
14.	PACKAGE=awswrangler  
15.	  
16.	# Note that "base" is special environment name, include it there as well.  
17.	for env in base /home/ec2-user/anaconda3/envs/*; do  
18.	    source /home/ec2-user/anaconda3/bin/activate $(basename "$env")  
19.	    if [ $env = 'JupyterSystemEnv' ]; then  
20.	        continue  
21.	    fi  
22.	    nohup pip install --upgrade "$PACKAGE" &  
23.	    source /home/ec2-user/anaconda3/bin/deactivate  
24.	done  
25.	EOF 

1.	[  
2.	  {  
3.	     "Classification": "spark-env",  
4.	     "Configurations": [  
5.	       {  
6.	         "Classification": "export",  
7.	         "Properties": {  
8.	            "PYSPARK_PYTHON": "/usr/bin/python3"  
9.	          }  
10.	       }  
11.	    ]  
12.	  }  
13.	]  

1.	#!/usr/bin/env bash  
2.	set -ex  
3.	  
4.	sudo pip-3.6 install awswrangler