数据技术与数据科学：基于 Batch 云计算的遥感数据计算实验

import json
import boto3

dynamodb_client = boto3.client('dynamodb')

def lambda_handler(event, context):
table_name = 'task-log'
existing_tables = dynamodb_client.list_tables()['TableNames']

if table_name not in existing_tables:
response = dynamodb_client.create_table(
AttributeDefinitions=[
{
'AttributeName': 'TaskID',
'AttributeType': 'S'
}
],
TableName='task-log',
KeySchema=[
{
'AttributeName': 'TaskID',
'KeyType': 'HASH'
}
],
BillingMode='PAY_PER_REQUEST')
print(response)
print(event)
return event

sudo yum update -y
sudo yum install docker
sudo service docker start

mkdir combands

boto3

# -*- coding: utf-8 -*-
import os
from osgeo import gdal
import boto3
import time

def input_log(files,date,success):
dynamodb_client = boto3.resource('dynamodb',region_name=region)
table = dynamodb_client.Table('task-log')
exec_time=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
TaskID=str(files)+"-"+str(date)
response = table.put_item(
Item={
'TaskID': TaskID,
'files': files,
'date': date,
'exec_time': exec_time,
'result':success
}
)
return response

env_dist = os.environ

print(env_dist.get('files'))
print(env_dist.get('date'))
print(env_dist.get('output_bucket'))
print(env_dist.get('region_name'))

region=env_dist.get('region_name')
files=env_dist.get('files')
strdate = env_dist.get('date')
bucketname=files.split('//')[1].split('/')[0]
objdir='/'.join(files.split('//')[1].split('/')[1:])
output_bucket=env_dist.get('output_bucket')

s3_client = boto3.client('s3',region_name=region)
s3_client.download_file(Bucket=bucketname,Key=objdir+'B02.jp2',Filename='B02.jp2',ExtraArgs={'RequestPayer':'requester'})
s3_client.download_file(Bucket=bucketname,Key=objdir+'B03.jp2',Filename='B03.jp2',ExtraArgs={'RequestPayer':'requester'})
s3_client.download_file(Bucket=bucketname,Key=objdir+'B04.jp2',Filename='B04.jp2',ExtraArgs={'RequestPayer':'requester'})
print('download files successful')

#os.chdir(r'D:\')
band1_fn = 'B02.jp2'
band2_fn = 'B03.jp2'
band3_fn = 'B04.jp2'

outputfile='sentinel_l2a_'+strdate+'_RGB.tif'

in_ds = gdal.Open(band1_fn)
#print(in_ds)
in_band = in_ds.GetRasterBand(1)
print('*******************************')
#print(in_band)
gtiff_driver = gdal.GetDriverByName('Gtiff')

out_ds = gtiff_driver.Create(outputfile,in_band.XSize,in_band.YSize,3,in_band.DataType)
out_ds.SetProjection(in_ds.GetProjection())
out_ds.SetGeoTransform(in_ds.GetGeoTransform())

in_data = in_band.ReadAsArray()
out_band = out_ds.GetRasterBand(3)
out_band.WriteArray(in_data)

in_ds = gdal.Open(band2_fn)
out_band = out_ds.GetRasterBand(2)
out_band.WriteArray(in_ds.ReadAsArray())

out_ds.GetRasterBand(1).WriteArray(gdal.Open(band3_fn).ReadAsArray())

out_ds.FlushCache()
for i in range(1,4):
out_ds.GetRasterBand(i).ComputeStatistics(False)

out_ds.BuildOverviews('average',[2,4,8,16,32])

del out_ds

#upload result data to S3
s3_client.upload_file("./"+outputfile,output_bucket,strdate+'/'+outputfile)
print('upload files successful')

#log result into ddb log table
log_resp = input_log(files,strdate,"success")
print("input log successful")
print('Task Finished')

vim Dockerfile
FROM ubuntu:trusty
COPY requirements.txt /tmp/requirements.txt
# Install required software via apt and pip
RUN sudo apt-get -y update && \
apt-get install -y \
awscli \
python \
python3-pip \
software-properties-common \
&& add-apt-repository ppa:ubuntugis/ppa \
&& apt-get -y update \
&& apt-get install -y gdal-bin \
&& apt-get install -y python3-gdal \
&& apt-get install -y python3-numpy \
&& pip3 install --requirement /tmp/requirements.txt
# Copy the script to Docker image and add execute permissions
COPY combands.py combands.py
RUN chmod +x combands.py

sudo su

{
"Comment": "Data process flow",
"StartAt": "Redefine task",
"States": {
"Redefine task": {
"Type": "Task",
"Resource": "arn:aws:states:::lambda:invoke",
"ResultPath": null,
"Parameters": {
"FunctionName": "<此处使用 lambda arn 替换>",
"Payload": {
"Input.$": "$"
}
},
"Next": "Map state"
},
"Map state": {
"Type": "Map",
"MaxConcurrency": 0,
"InputPath": "$.detail",
"ItemsPath": "$.cells",
"Iterator": {
"StartAt": "batch compute",
"States": {
"batch compute": {
"Type": "Task",
"Resource": "arn:aws:states:::batch:submitJob.sync",
"Parameters": {
"JobDefinition": "<此处使用Batch 作业定义arn 替换>",
"JobName": "combands",
"JobQueue": "<此处使用Batch 作业队列arn替换>",
"ContainerOverrides": {
"Command": [
"/usr/bin/python3",
"combands.py"
],
"Environment": [{
"Name": "files",
"Value.$": "$.files"
},
{
"Name": "date",
"Value.$": "$.date"
},
{
"Name": "output_bucket",
"Value.$": "$.output_bucket"
},
{
"Name": "region_name",
"Value.$": "$.region_name"
}
]
}
},
"End": true,
"Catch": [{
"ErrorEquals": [
"States.ALL"
],
"Next": "Failure Notify"
}]
},
"Failure Notify": {
"Type": "Task",
"Resource": "arn:aws:states:::sns:publish",
"Parameters": {
"TopicArn": "<此处使用sns topic arn 替换>",
"Message": {
"Input": "SubTask Failure!"
}
},
"End": true
}
}
},
"Next": "Task Complete"
},
"Task Complete": {
"Type": "Task",
"Resource": "arn:aws:states:::sns:publish",
"Parameters": {
"TopicArn": "<此处使用sns topic arn 替换>",
"Message": {
"Input": "Task have complete!"
}
},
"End": true
}
}
}

{
"detail": {
"cells": [
{ "files":"s3://sentinel-s2-l2a/tiles/48/R/XT/2020/8/1/0/R10m/","date":"20200801","output_bucket":"<前面步骤中新建的bucket名称>","region_name":"<实验所在的region 如：ap-northeast-1 >"},
{ "files":"s3://sentinel-s2-l2a/tiles/48/R/XT/2020/8/4/0/R10m/","date":"20200804","output_bucket":"<前面步骤中新建的bucket名称>","region_name":"<实验所在的region 如：ap-northeast-1 >"}
]
}
}