Интеграция AWS Lambda и Snowflake + автоматизация с помощью Snowpipe | ETL | Пакет развертывания ... скачать в хорошем качестве

Интеграция AWS Lambda и Snowflake + автоматизация с помощью Snowpipe | ETL | Пакет развертывания ...

В этом проекте лямбда-функция (написанная на Python) будет срабатывать при поступлении новых данных в определённый контейнер S3. Функция извлечёт CSV-данные из фрейма данных Pandas, обработает или преобразует их в соответствии с бизнес-логикой и загрузит данные в другой контейнер S3. После этого Snowpipe автоматически загрузит новые данные в целевую таблицу для анализа или визуализации (эту таблицу Snowpipe можно подключить к любому инструменту визуализации). Настройка подсистемы Bash и Linux для Windows 10    • Windows 10 Bash & Linux Subsystem Setup   Команды: python3 --version pip3 install pandas==1.0.3 -t . pip3 install boto3==1.13.11 -t . pip3 install s3fs==0.4.2 -t . Код Python: импорт pandas как pd импорт boto3 импорт io из io импорт StringIO def lambda_handler(event, context): s3_file_key = event['Records'][0]['s3']['object']['key']; bucket = 'sourcefiledemo'; s3 = boto3.client('s3', aws_access_key_id='', aws_secret_access_key='') obj = s3.get_object(Bucket=bucket, Key=s3_file_key) initial_df = pd.read_csv(io.BytesIO(obj['Body'].read())); имя_сервиса = 's3' имя_региона = 'us-east-2' aws_access_key_id = '' aws_secret_access_key = '' s3_resource = boto3.resource( имя_сервиса = имя_сервиса, имя_региона = имя_региона, aws_access_key_id = идентификатор_ключа_aws_access_key, aws_secret_access_key = ключ_aws_secret_access_key ) bucket = 'destinationbucketdemoshow'; df = initial_df[(initial_df.species == "setosa")]; csv_buffer = StringIO() df.to_csv(csv_buffer,index=False); s3_resource.Object(bucket, s3_file_key).put(Body=csv_buffer.getvalue()) Код снежинки: удалить базу данных s3_to_snowflake; --Создание базы данных создать базу данных, если её нет, s3_to_snowflake; использовать s3_to_snowflake; --Создание таблицы создать или заменить таблицу s3_to_snowflake.PUBLIC.Iris_dataset (sepal_length number(10,5), sepal_width number(10,4), petal_length number(10,3), petal_width number(10,4), species varchar(20)); Создать или заменить формат файла my_csv_format type = csv field_delimiter = ',' skip_header = 1 field_optionally_enclosed_by = '"' null_if = ('NULL', 'null') empty_field_as_null = true; Создать или заменить этап s3_to_snowflake.PUBLIC.Snow_stage url="s3://" credentials=(aws_key_id='' aws_secret_key='') file_format = my_csv_format; list @Snow_stage; Создать или заменить канал s3_to_snowflake.PUBLIC.spa auto_ingest=true как скопировать в s3_to_snowflake.PUBLIC.Iris_dataset из @s3_to_snowflake.PUBLIC.Snow_stage ФОРМАТ_ФАЙЛА=(ИМЯ_ФОРМАТА=my_csv_format); показать каналы; выбрать * из s3_to_snowflake.PUBLIC.Iris_dataset; выбрать количество(*) из s3_to_snowflake.PUBLIC.Iris_dataset; удалить из s3_to_snowflake.PUBLIC.Iris_dataset; #Снежинка #снежная труба #python #etl