当前位置：首页 > news >正文

承德网站制作加盟小程序模板

news 2025/8/10 21:24:05

承德网站制作加盟,小程序模板,建设工程施工合同（示范文本）,江苏省常州建设高等职业技术学校网站本文是一份开箱即用的全自动测试脚本，用于在 EMR Serverless 上提交一个 Delta Lake 作业。本文完全遵循《最佳实践：如何优雅地提交一个 Amazon EMR Serverless 作业？》一文给出的标准和规范！ 1. 导出环境相关变量注意&#x…

本文是一份开箱即用的全自动测试脚本，用于在 EMR Serverless 上提交一个 Delta Lake 作业。本文完全遵循《最佳实践：如何优雅地提交一个 Amazon EMR Serverless 作业？》一文给出的标准和规范！

1. 导出环境相关变量

注意：以下仅为示意值，实操时请根据个人环境替换相关值。

export APP_NAME="emr-serverless-deltalake-test"
export APP_S3_HOME="s3://$APP_NAME"
export APP_LOCAL_HOME="/home/ec2-user/$APP_NAME"
export EMR_SERVERLESS_APP_ID='00fbfel40ee59k09'
export EMR_SERVERLESS_EXECUTION_ROLE_ARN='arn:aws:iam::1111111111111:role/EMR_SERVERLESS_ADMIN'

2. 创建作业专属工作目录和S3存储桶

mkdir -p $APP_LOCAL_HOME
aws s3 mb $APP_S3_HOME

3. 准备作业脚本

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.py

4. 准备作业描述文件

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.class=com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension --conf spark.sql.catalog.spark_catalog=org.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.json

5. 提交 & 监控作业

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[0m"sleep 1doneelseecho -ne "The job [ $EMR_SERVERLESS_JOB_RUN_ID ] is [ $jobStatus ]\n\n"breakfi
done

6. 检查错误

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 --color=always -r -i -E 'error|failed|exception' $JOB_LOG_HOME