网站开发培训课程,广东网站建设公司,有做喜糖的网站吗,1千元以下做网站的公司本文是一份开箱即用的全自动测试脚本#xff0c;用于在 EMR Serverless 上提交一个 Delta Lake 作业。本文完全遵循《最佳实践#xff1a;如何优雅地提交一个 Amazon EMR Serverless 作业#xff1f;》 一文给出的标准和规范#xff01;
1. 导出环境相关变量
注意#x…本文是一份开箱即用的全自动测试脚本用于在 EMR Serverless 上提交一个 Delta Lake 作业。本文完全遵循《最佳实践如何优雅地提交一个 Amazon EMR Serverless 作业》 一文给出的标准和规范
1. 导出环境相关变量
注意 以下仅为示意值实操时请根据个人环境替换相关值。
export APP_NAMEemr-serverless-deltalake-test
export APP_S3_HOMEs3://$APP_NAME
export APP_LOCAL_HOME/home/ec2-user/$APP_NAME
export EMR_SERVERLESS_APP_ID00fbfel40ee59k09
export EMR_SERVERLESS_EXECUTION_ROLE_ARNarn:aws:iam::1111111111111:role/EMR_SERVERLESS_ADMIN2. 创建作业专属工作目录和S3存储桶
mkdir -p $APP_LOCAL_HOME
aws s3 mb $APP_S3_HOME3. 准备作业脚本
cat EOF $APP_LOCAL_HOME/delta_table.py
from datetime import datetime
from pyspark import SparkConf, SparkContext
from pyspark.sql import HiveContext, SparkSessionspark SparkSession\.builder\.appName(Delta-Lake integration demo - create tables)\.enableHiveSupport()\.getOrCreate()## Create a DataFrame
data spark.createDataFrame([(100, 2015-01-01, 2015-01-01T13:51:39.340396Z),
(101, 2015-01-01, 2015-01-01T12:14:58.597216Z),
(102, 2015-01-01, 2015-01-01T13:51:40.417052Z),
(103, 2015-01-01, 2015-01-01T13:51:40.519832Z)],
[id, creation_date, last_update_time])spark.sql(drop table if exists delta_table)## Write a DataFrame as a Delta Lake dataset to the S3 location
spark.sql(CREATE TABLE IF NOT EXISTS delta_table (id string, creation_date string,
last_update_time string)
USING delta location
s3://$APP_NAME/delta_table);data.writeTo(delta_table).append()
EOF
aws s3 cp $APP_LOCAL_HOME/delta_table.py $APP_S3_HOME/delta_table.py4. 准备作业描述文件
cat EOF $APP_LOCAL_HOME/start-job-run.json
{name:$APP_NAME,applicationId:$EMR_SERVERLESS_APP_ID,executionRoleArn:$EMR_SERVERLESS_EXECUTION_ROLE_ARN,jobDriver:{sparkSubmit:{entryPoint:s3://$APP_NAME/delta-test.py,sparkSubmitParameters:--conf spark.hadoop.hive.metastore.client.factory.classcom.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory --conf spark.sql.extensionsio.delta.sql.DeltaSparkSessionExtension --conf spark.sql.catalog.spark_catalogorg.apache.spark.sql.delta.catalog.DeltaCatalog --conf spark.jars/usr/share/aws/delta/lib/delta-core.jar,/usr/share/aws/delta/lib/delta-storage.jar,/usr/share/aws/delta/lib/delta-storage-s3-dynamodb.jar}},configurationOverrides:{monitoringConfiguration:{s3MonitoringConfiguration:{logUri:$APP_S3_HOME/logs}}}
}
EOF
jq . $APP_LOCAL_HOME/start-job-run.json5. 提交 监控 作业
export EMR_SERVERLESS_JOB_RUN_ID$(aws emr-serverless start-job-run \--no-paginate --no-cli-pager --output text \--name apache-hudi-delta-streamer \--application-id $EMR_SERVERLESS_APP_ID \--execution-role-arn $EMR_SERVERLESS_EXECUTION_ROLE_ARN \--execution-timeout-minutes 0 \--cli-input-json file://$APP_LOCAL_HOME/start-job-run.json \--query jobRunId) \
now$(date %s)sec \
while true; dojobStatus$(aws emr-serverless get-job-run \--no-paginate --no-cli-pager --output text \--application-id $EMR_SERVERLESS_APP_ID \--job-run-id $EMR_SERVERLESS_JOB_RUN_ID \--query jobRun.state)if [ $jobStatus PENDING ] || [ $jobStatus SCHEDULED ] || [ $jobStatus RUNNING ]; thenfor i in {0..5}; doecho -ne \E[33;5m The job [ $EMR_SERVERLESS_JOB_RUN_ID ] state is [ $jobStatus ], duration [ $(date -u --date now-$now %H:%M:%S) ] ....\r\E[0msleep 1doneelseecho -ne The job [ $EMR_SERVERLESS_JOB_RUN_ID ] is [ $jobStatus ]\n\nbreakfi
done6. 检查错误
JOB_LOG_HOME$APP_LOCAL_HOME/log/$EMR_SERVERLESS_JOB_RUN_ID
rm -rf $JOB_LOG_HOME mkdir -p $JOB_LOG_HOME
aws s3 cp --recursive $APP_S3_HOME/logs/applications/$EMR_SERVERLESS_APP_ID/jobs/$EMR_SERVERLESS_JOB_RUN_ID/ $JOB_LOG_HOME /dev/null
gzip -d -r -f $JOB_LOG_HOME /dev/null
grep --coloralways -r -i -E error|failed|exception $JOB_LOG_HOME
