AWS使用笔记 Sagemaker

AWS 全称Amazon web service(亚马逊网络服务)，是亚马逊公司旗下云计算服务平台，为全世界各个国家和地区的客户提供一整套基础设施和云解决方案。
AWS面向用户提供包括弹性计算、存储、数据库、物联网在内的一整套云计算服务，帮助企业降低IT投入和维护成本，轻松上云
从概念是来看，AWS提供了一系列的托管产品，帮助我们在没有物理服务器的情况下，照样可以正常完成软件开发中的各种需求，也就是我们常说的云服务。

AWS使用学习笔记

Sagemaker操作S3文件: 列出S3文件列表

import os
# 递归列出文件夹及文件，recursive=True
def list_s3_dir(s3_uri, recursive=False):
    cmd = f'aws s3 ls {s3_uri} --recursive' if recursive else f'aws s3 ls {s3_uri}'
    result = os.popen(cmd)
    res_list = result.readlines()
    r = []
    for x in res_list:
        file = x[x.rfind(' ')+1:-1]
        r.append(file)
    result.close()
    return r

r = list_s3_dir("s3://xxxxx/", True)
print(r)

Sagemaker操作S3文件: 删除S3文件

import os
def rm_s3_dir(s3_uri, recursive=False):
    """
     删除文件，recursive=False
     删除文件夹 recursive=True
    """
    cmd = f'aws s3 rm {s3_uri} --recursive' if recursive else f'aws s3 rm {s3_uri}'
    result = os.system(cmd)
    return result

Sagemaker操作Athena数据库 将df上传到数据表

import datetime
import awswrangler as wr

# 排除时区造成的时间差
def datetime_beijing(datetime_):
    beijing_time = datetime_ + datetime.timedelta(hours=8)
    return beijing_time

# 将df存为分区表
def gen_part_parquet(df: str,path: str,part: list,table: str,dtype: dict):
    if len(df)>0:
        print(datetime_beijing(datetime.datetime.now()), '###### generate {type} parquet start...'.format(type=table))
        wr.s3.to_parquet(
                df = df, path= path, dataset=True, mode="overwrite_partitions", partition_cols=part, sanitize_columns=True,
                database="xxx", table=table, dtype=dtype
            )
        print(datetime_beijing(datetime.datetime.now()), '###### generate {type} parquet end...'.format(type=table))
    else:
        print(df," is empty....")

# 将df存为非分区表
def gen_s3_parquet(df: str,path: str,table: str, dtype=None):
    if len(df)>0:
        print(datetime_beijing(datetime.datetime.now()), '###### generate {type} start...'.format(type=table))
        wr.s3.to_parquet(
                df = df, path= path, dataset=True, mode="overwrite",
                sanitize_columns=True, database="xxx", table=table, dtype=dtype
            )
        print(datetime_beijing(datetime.datetime.now()), '###### generate {type} end...'.format(type=table))
    else:
        print(df," is empty....")