互联网行业每天都有大量的日志生成，需要在固定时间段对数据进行 ETL 工作。用户常规的做法是启动一组长期运行的 EMR 集群，配置远程提交任务的服务器，结合自身的任务调度系统定期提交任务，但集群执行完成任务之后会闲置，造成不必要的开销。另一种方法是在需要执行任务的时候启动集群，任务完成之后关闭集群，但因为每次启动集群后，主节点与核心节点的 IP 都会发生分变化，导致每次都需要重新配置提交任务的服务器，造成额外的工作负担。本文介绍了一种通过 Apache Airflow 任务调度系统动态启停 Amazon EMR 集群的方法，并通过 EMR 内置的 Livy 远程提交作业，这样可以节省大量的成本并且无需进行过多的额外配置。
1. 相关技术介绍
在开始之前，请先对以下技术进行简单了解。
1.1 Apache Airflow
Apache Airflow 是一款开源的任务调度系统，用户通过创建 DAG（有向无环图）来定义任务的流程，图中的每个节点就是需要执行的任务，不同 DAG 之间的任务可以相互依赖。通过 Airflow 我们可以定时执行脚本，并且它提供了 web 界面供用户可视化地查看任务执行情况。
1.2 Apache Livy
Apache Livy 是 Hadoop 生态圈中提供远程提交任务功能的应用程序。它以 Rest API 的方式提供了 Session 与 Batches 两种集群执行任务的方法。Session 指的是将集群需要执行的代码写在对 Livy 请求中，目前支持 spark、pyspark、sparkr 与 sql 等四种方式与集群交互。Bathches 指的是将代码存放在指定位置，在请求中提供路径，让集群执行代码。例如将 jar 包存放在 S3 上，在请求 Livy 的时候提供 jar 包的路径，从而让集群直接执行 jar 包，好处是无需在集群上配置执行代码所需的依赖。
2. 演练
通过本文示例，我将向您展示如何实现以下方案：
基于开源调度工具 Airflow 编排提交 Spark Jobs 到 EMR 做批处理，Job 开始之前启动 EMR 集群，对集群节点采用 Spot 实例，所有 Job 结束后关闭 EMR 集群。
2.1 流程架构图与过程简介
（1）在一台 EC2 上配置 Airflow；
（2）定义 Airflow 工作流，其中包含创建集群，Spark 任务步骤与终止集群等任务；
（3）向 Livy 提交任务；
（4）EMR 从 S3 中读取数据，对数据进行处理完成之后重新写入 S3；
（5）工作完成，终止集群。
2.2 前提条件
（1）本文示例所使用的区域 us-east-1；
（2）在该区域创建一台 EC2，并确保与 EC2 绑定的 IAM Role 有 EMR 集群的 Full Access；
（3）拥有创建 EMR 集群所需的默认角色：EMR_DefaultRole 与 EMR_EC2_DefaultRole；
（4）创建 S3 桶，下载 jar 包 spark-examples_2.11-2.4.4 和数据集 emrdata.txt，并上传至 s3。
2.3 实现过程
2.3.1 在 EC2 上配置 Airflow
（1）登陆 EC2，安装 Airflow 已经相关依赖
Python
sudo yum update -ysudo yum install -y python-pip gcc mysql-devel python-devel mysqlsudo pip install mysql-pythonsudo yum install -y python3sudo pip3 install boto3sudo pip3 install requests
# 安装Airflowsudo pip install apache-airflowsudo pip install 'apache-airflow[celery]'airflow initdb
复制代码
（2）创建 RDS for Mysql 数据库供 Airflow 使用，对数据库性能要求不高，因此使用默认配置即可
（3）更改 airflow.cfg 配置文件，并测试是否能打开 Airflow 的 web 页面
Python
cd airflowvim airflow.cfg
# 找到sql_alchemy_conn等参数所在位置，替换为创建的数据库信息sql_alchemy_conn = mysql://admin:12345678@database-for-airflow.cdtwa5j4xten.us-east-1.rds.amazonaws.com/airflowdb
# Exit vim, Update Airflow Databaseairflow initdb
# 配置celery相关参数vim airflow.cfg
# 找到executor位置，将执行器设置为celery，可保证不相互依赖的任务可以并行执行executor = CeleryExecutor
# 找到broker_url与result_backend参数的位置broker_url = sqla+mysql://admin:12345678@database-for-airflow.cdtwa5j4xten.us-east-1.rds.amazonaws.com:3306/airflowdbresult_backend = db+mysql://admin:12345678@database-for-airflow.cdtwa5j4xten.us-east-1.rds.amazonaws.com:3306/airflowdb
# 开启airflow的webserver，在网页上输入EC2的DNS，查看是否能打开网页（注意打开安全组，并且如果本地连上的是公司的vpn，可能会出现无法打开网页的情况）airflow webserver -p 8080 &
# 启动workerairflow worker &
# 启动flower，可对worker中的任务进行可视化(要看到网页注意打开5555端口)airflow flower &
复制代码
复制代码
2.3.2 定义工作流
现定义如下两个 Airflow 的 DAG：
dag_transform_calpi
（1）create_emr_cluster：创建 EMR 集群；
Python
# -*- coding: UTF-8 -*-
import boto3import time
emr_client = boto3.client('emr', region_name='us-east-1')
# 定义集群名称，集群名称不要与当前运行的集群重名name = 'emr-cluster'
# 定义instance，可自定义实例的数量与类型intances = { 'InstanceGroups': [ { 'Market': 'SPOT', 'InstanceRole': 'MASTER', 'InstanceType': 'm4.xlarge', 'InstanceCount': 1, }, { 'Market': 'SPOT', 'InstanceRole': 'CORE', 'InstanceType': 'm4.xlarge', 'InstanceCount': 2, } ], 'KeepJobFlowAliveWhenNoSteps': True}
# 定义集群中的应用applications = [ { 'Name': 'Hadoop' }, { 'Name': 'Pig' }, { 'Name': 'Livy' }, { 'Name': 'Hive' }, { 'Name': 'Hue' }, { 'Name': 'Spark' }]
if __name__ == '__main__':
# 创建emr集群 emr_client.run_job_flow( Name=name, ReleaseLabel='emr-5.12.0', Instances=intances, Applications=applications, JobFlowRole='EMR_EC2_DefaultRole', ServiceRole='EMR_DefaultRole')
# 持续发送请求，直到创建的集群状态处于Waiting为止 flag = True while flag: time.sleep(20) r = emr_client.list_clusters(ClusterStates=['WAITING']) for i in r['Clusters']: if i['Name'] == name: flag = False
复制代码
（2）create_livy_session：创建 Livy 会话；
Python
# -*- coding: UTF-8 -*-import requestsimport jsonimport pprintimport boto3
# 获取集群的DNS，其中name为你的集群名称name = 'emr-cluster'emr_client = boto3.client('emr', region_name='us-east-1')r = emr_client.list_clusters(ClusterStates=['WAITING'])for i in r['Clusters']: if i['Name'] == name: cluster_id = i['Id']r = emr_client.describe_cluster(ClusterId=cluster_id)emr_dns = r['Cluster']['MasterPublicDnsName']
# livy_host为配置在emr集群上livy的url，无需修改代码livy_host = 'http://' + emr_dns + ':8998'data = {'kind': 'pyspark'}headers = {'Content-Type': 'application/json'}r = requests.post(livy_host + '/sessions', data=json.dumps(data), headers=headers)pprint.pprint(r.json())
复制代码
（3）sleep：等待会话创建完成；
（4）calpi：以 batches 的方式执行 spark 任务计算 pi 值；
Python
# -*- coding: UTF-8 -*-import requestsimport jsonimport textwrapimport pprintimport boto3
# 获取执行jar包任务的livy batch的url，其中name为你的集群名称name = 'emr-cluster'emr_client = boto3.client('emr', region_name='us-east-1')r = emr_client.list_clusters(ClusterStates=['WAITING'])for i in r['Clusters']: if i['Name'] == name: cluster_id = i['Id']r = emr_client.describe_cluster(ClusterId=cluster_id)emr_dns = r['Cluster']['MasterPublicDnsName']batch_url = 'http://' + emr_dns + ':8998/batches'headers = {'Content-Type': 'application/json'}
# 提交任务data = {"file": "s3://xiaoyj/emr/spark-examples_2.11-2.4.4.jar", "className": "org.apache.spark.examples.SparkPi"}r = requests.post(batch_url, data=json.dumps(data), headers=headers)pprint.pprint(r.json())
复制代码
（5）query_completed：外部任务，依赖于第二个 DAG（dag_query），即等待查询完成之后，执行下一个任务；
（6）终止集群。
Python
# -*- coding: UTF-8 -*-import boto3import time
# 终止集群，其中name为你的集群名称name = 'emr-cluster'emr_client = boto3.client('emr', region_name='us-east-1')flag = Truewhile flag: time.sleep(120) r = emr_client.list_clusters(ClusterStates=['WAITING']) for i in r['Clusters']: if i['Name'] == name: emr_client.terminate_job_flows(JobFlowIds=[i['Id']]) flag = False
复制代码
dag_query
（1）sleep_completed：外部任务，依赖于第一个 DAG（dag_transform_calpi），即等待 Livy 会话执行下一个任务；
（2）transform：对之前上传到 S3 上的文本文件进行聚合、转换；
Python
# -*- coding: UTF-8 -*-
import requestsimport jsonimport textwrapimport pprintimport boto3
# 获取提交任务的livy_url，其中name为你的集群名称name = 'emr-cluster'emr_client = boto3.client('emr', region_name='us-east-1')r = emr_client.list_clusters(ClusterStates=['WAITING'])for i in r['Clusters']: if i['Name'] == name: cluster_id = i['Id']r = emr_client.describe_cluster(ClusterId=cluster_id)emr_dns = r['Cluster']['MasterPublicDnsName']livy_url = 'http://' + emr_dns + ':8998/sessions/0/statements'headers = {'Content-Type': 'application/json'}
# 提交任务，data中的code为在emr中执行的代码，对s3中的文件进行转化操作，完成后将结果存放回s3作为中间结果data = { 'code': textwrap.dedent(""" import json sc._jsc.hadoopConfiguration().set('fs.s3a.endpoint', 's3-us-east-2.amazonaws.com') text_file = sc.textFile("s3a://xiaoyj/emr/emrdata.txt") text_file = text_file.map(lambda x: x.split('::')) text_file = text_file.map(lambda x: (int(x[0]), x[1:])) text_file = text_file.groupByKey().map(lambda x: (x[0], list(x[1]))) text_file = text_file.sortByKey() text_file = text_file.map(lambda x: {x[0]: x[1]}) text_file = text_file.map(lambda x: json.dumps(x)) text_file.coalesce(1).saveAsTextFile("s3a://xiaoyj/emr/middle_result") print("Transform Complete!") """)}r = requests.post(livy_url, data=json.dumps(data), headers=headers)pprint.pprint(r.json())
复制代码
（3）check_s3：检查 S3 中是否有上一步生成的中间结果；
Python
# -*- coding: UTF-8 -*-
import boto3import time
# 轮询s3，确认transform任务是否执行完成（即s3中是否有middle_result文件生成），name为你的s3桶名称name = 'xiaoyj's3_client = boto3.client('s3', region_name='us-east-1')flag = Truewhile flag: time.sleep(60) r = s3_client.list_objects(Bucket=name) for i in r['Contents']: if i['Key'] == 'emr/middle_result/part-00000': flag = False
复制代码
（4）query：对上一步生成的中间结果进行查询。
Python
# -*- coding: UTF-8 -*-
import requestsimport jsonimport textwrapimport pprintimport boto3
# 获取提交任务的livy_url，其中name为你的集群名称name = 'emr-cluster'emr_client = boto3.client('emr', region_name='us-east-1')r = emr_client.list_clusters(ClusterStates=['WAITING'])for i in r['Clusters']: if i['Name'] == name: cluster_id = i['Id']r = emr_client.describe_cluster(ClusterId=cluster_id)emr_dns = r['Cluster']['MasterPublicDnsName']livy_url = 'http://' + emr_dns + ':8998/sessions/0/statements'headers = {'Content-Type': 'application/json'}
# 提交任务，data中的code为在emr中执行的代码，对s3中的文件进行转化操作，完成后将结果存放回s3作为中间结果data = { 'code': textwrap.dedent(""" import json from pyspark.sql import HiveContext, Row hiveCtx = HiveContext(sc) input = hiveCtx.read.json("s3a://xiaoyj/emr/middle_result/part-00000") input.registe rTempTable("tbn") result = hiveCtx.sql("SELECT size(`9`) from tbn") result = result.rdd.map(lambda row: row[0]) result.coalesce(1).saveAsTextFile("s3a://xiaoyj/emr/result") print("Search Complete!") """)}r = requests.post(livy_url, data=json.dumps(data), headers=headers)pprint.pprint(r.json())
复制代码
2.3.3 创建 Airflow 工作流
（1）在 airflow 文件夹中创建 dags 文件夹，并进入到文件夹中；
（2）定义工作流（注意开头的 # — coding: UTF-8 –不要省略，并且 bash_command 需替换为自己任务所在的路径）；
Python
vim dag_transform_calpi.py
# -*- coding: UTF-8 -*-
from airflow import DAGfrom airflow.operators.bash_operator import BashOperatorfrom datetime import datetime, timedeltafrom airflow.sensors.external_task_sensor import ExternalTaskSensor
default_args = { 'owner': 'Airflow', 'depends_on_past': False, 'start_date': datetime.now().replace(microsecond=0), 'email': ['756044579@qq.com'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 0, 'retry_delay': timedelta(minutes=5), # 'queue': 'bash_queue', # 'pool': 'backfill', # 'priority_weight': 10, # 'end_date': datetime(2016, 1, 1),}
dag = DAG('dag_transform_calpi', default_args=default_args, schedule_interval=timedelta(days=1))
# 创建emr集群t0 = BashOperator( task_id='create_emr_cluster', bash_command='python3 /Users/xiaoyj/Desktop/emr_poc/create_emr_cluster.py', dag=dag)
# 创建livy的会话t1 = BashOperator( task_id='create_livy_session', bash_command='python3 /Users/xiaoyj/Desktop/emr_poc/create_session.py', dag=dag)
# 等待会话创建完成t2 = BashOperator( task_id='sleep', bash_command='sleep 20', dag=dag)
# 计算pi值t3 = BashOperator( task_id='calpi', bash_command='python3 /Users/xiaoyj/Desktop/emr_poc/calpi.py', dag=dag)
# 终止emr集群t4 = BashOperator( task_id='terminate_cluster', bash_command='python3 /Users/xiaoyj/Desktop/emr_poc/terminate_cluster.py', dag=dag)
# dag_query中的spark sql任务external_task = ExternalTaskSensor( external_task_id='query', task_id='query_completed', external_dag_id='dag_query', dag=dag)
# 定义airflow的有向无环图t0 >> t1t1 >> t2t2 >> t3external_task >> t4
复制代码
Python
vim dag_query.py # -*- coding: UTF-8 -*-from airflow import DAGfrom airflow.operators.bash_operator import BashOperatorfrom datetime import datetime, timedeltafrom airflow.sensors.external_task_sensor import ExternalTaskSensor
default_args = { 'owner': 'Airflow', 'depends_on_past': False, 'start_date': datetime.now().replace(microsecond=0), 'email': ['756044579@qq.com'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 0, 'retry_delay': timedelta(minutes=5), # 'queue': 'bash_queue', # 'pool': 'backfill', # 'priority_weight': 10, # 'end_date': datetime(2016, 1, 1),}
dag = DAG('dag_query', default_args=default_args, schedule_interval=timedelta(days=1))
# 对s3上的文本文件进行转化操作t0 = BashOperator( task_id='transform', bash_command='python3 /Users/xiaoyj/Desktop/emr_poc/transform.py', dag=dag)
# 轮询s3，查看中间结果是否生成t1 = BashOperator( task_id='check_s3', bash_command='python3 /Users/xiaoyj/Desktop/emr_poc/check_s3.py', dag=dag)
# spark sql任务t2 = BashOperator( task_id='query', bash_command='python3 /Users/xiaoyj/Desktop/emr_poc/query.py', dag=dag)
# dag_transform_calpi中的sleep任务external_task = ExternalTaskSensor( external_task_id='sleep', task_id='sleep_completed', external_dag_id='dag_transform_calpi', dag=dag)external_task >> t0t0 >> t1t1 >> t2
复制代码
（3）重制Airflow数据库；
Python
复制代码
airflow resetdb
（4）启动Airflow，-s为当前日期，-e是结束日期，均设置为当日的日期（若工作流执行失败并想重头开始执行工作，需要先执行airflow resetdb）
Python
复制代码
airflow backfill dag_transform_calpi -s 2019-12-02 -e 2019-12-02 & airflow backfill dag_query -s 2019-12-02 -e 2019-12-02
## 3. 展示
（1）打开AWS EMR控制台，可以观察到集群正在创建；
[](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy5.png)](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy5.jpg)
（2）待集群创建完成后，获取主节点DNS，并打开网页；
[](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy6.png)](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy6.jpg)
（3）观察到Livy上并行提交了两个任务分别是spark对文本的tansform操作和jar包计算pi值的任务；
[](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy7.png)](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy7.jpg)
（4）pi值计算完成；
[](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy8.png)](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy8.jpg)
（5）待Transform任务完成，Spark SQL任务开始执行；
[](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy9.png)](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy9.jpg)
（6）执行完成后可以在s3上可以看到Transform任务生成的middle result和Spark SQL任务生成的最终结果；
[](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy10.png)](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy10.jpg)
（7）下载middle_result中的文件，可以看到聚合结果；
[](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy11.png)](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy11.jpg)
（8）下载result中的文件，可以查看到最终结果（统计编号为9的列表中包含53组数据，-1表示其他json文件没有编号为9的；
[](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy12.png)](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy12.jpg)
（9）任务执行完毕，发现集群自动终止；
[](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy13.png)](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy13.jpg)
（10）再查看远程服务器上Airflow的web界面，发现两个dag已经执行完毕。
[](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy14.png)](https://s3.cn-north-1.amazonaws.com.cn/awschinablog/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy14.jpg)
## 4. 总结
本文展现了如何使用Airflow启动EMR集群，并通过Livy远程提交任务，在任务完成后终止集群。成本节省主要体现在两个方面：1）每天在需要执行ETL工作时启动集群，任务执行完成后终止集群，因此不会出现空闲的集群；2）EMR可以配合Spot实例使用，从而节省更多的成本。另一个好处是使用Livy无需额外配置远程提交任务的服务器，并且EMR集成了Livy的一键安装，造成了极大的方便。
## 本篇作者
<footer> 
### [](https://amazonaws-china.com/cn/blogs/china/tag/%E8%82%96%E5%85%83%E5%90%9B/)
AWS解决方案架构师，负责基于AWS云计算方案的架构咨询和设计实现，同时致力于数据分析与AI的研究与应用。</footer>
**作者介绍：**翟羽翔，AWS解决方案架构师，负责基于AWS云计算方案的架构咨询和设计实现，同时致力于数据湖的应用和推广。
**本文转载自AWS技术博客。**
**原文链接：**https://amazonaws-china.com/cn/blogs/china/dynamic-start-stop-of-emr-cluster-with-airflow-and-remote-submission-of-tasks-via-livy/复制代码
划线
评论
复制
更多内容推荐
SparkSQL 在有赞的实践
本文来自《2019年有赞技术大礼包》系列。
大数据基础：Spark 工作原理及基础概念
Apache Spark 是专为大规模数据处理而设计的快速通用计算引擎，在数据挖掘和机器学习领域有着广泛的应用，现在也已形成一个高速发展、应用广泛的生态系统。本文将为大家详细介绍 Spark 的核心技术原理。
如何用 Spark 计算引擎执行 FATE 联邦学习任务？
FATE 1.5 LTS 版本支持使用 Spark 作为底层的计算引擎，本文将对其实现细节以及使用进行简单介绍，方便用户在实际的使用过程中进行调优或者排查错误。
40｜命令式和声明式，谁才是驱动云原生的“引擎”？
这节课我们来聊聊命令式和声明式。
2023-03-10
使用 AWS Step Functions 和 AWS Glue 编排基于 Amazon Redshift 的 ETL 工作流（二）
Step Functions 编排
AWS 推出 Apache Airflow 全托管工作流 MWAA
最近，AWS推出了亚马逊Apache Airflow托管工作流(MWAA)，这是一项全托管的服务，简化了在AWS上运行开源版Apache Airflow和构建工作流来执行ETL作业和数据管道的工作。
多容器部署：如何利用 Docker Compose 快速搭建本地爬虫环境？
这节课，我们一起来学习如何使用 Docker Compose 来部署多个容器。
2023-02-07
31｜项目实战与部署：如何实现接口部署与访问？
在企业应用当中，把项目部署到服务器上，不但能让前端访问接口，也能供更多用户使用我们的平台。
2023-07-03
Hive 学习笔记（二）
本篇是Hive学习笔记第二篇，第一篇传送门https://xie.infoq.cn/article/0c0f67ab82ed951b27481341d
2021-07-07
39｜GitOps 最佳实践，ArgoCD 凭什么脱颖而出？
ArgoCD 能在众多 CD 工具中脱颖而出，除了 GitOps 大背景的推动以外，其自身也具备非常多优秀的特性。
2023-03-08
17｜组件监控：Kubernetes Node 组件的关键指标与数据采集
Kubernetes Node组件的关键指标与数据采集
2023-02-15
新的 Dataproc 可选组件支持 Apache Flink 和 Docker
集群。Flink。是一种广泛使用的容器技术。集群的每个节点。集群交互。中创建容器化的服务。集群。可移植性的更多信息。
教你如何在 Spark Scala/Java 应用中调用 Python 脚本
摘要：本文将介绍如何在 Spark scala 程序中调用 Python 脚本，Spark java程序调用的过程也大体相同。
2022-01-10
Spark :: 大画 Spark(前传)—Spark 的 Dispatcher 原理
在Spark的网络环境中，Netty是进行RPC通讯的底层通讯模块，在Netty之上，Spark构建起了一整套的通信发送与接收处理机制，这个机制涉及到了Spark的Dispatcher，MessageLoop，Endpoint，Inbox等一系列的概念和关系，理解诸如这些概念和关系是很多Spark初学者的
2021-08-28
Spark 任务等待与运行策略
前面我们提到了Spark的资源分配策略，资源配置有静态和动态两种模式，不同模式在任务提交后会有不同的内存占用行为，但是由于队列资源是有限的，因此会出现任务因为资源不够导致等待的情况。本节来详细分析一下任务提交后在的等待与运行影响因素。
2021-04-19
使用 AWS Step Functions 和 AWS Glue 编排基于 Amazon Redshift 的 ETL 工作流（一）
使用 AWS Glue Python Shell 构建
MRS 2.0 性能再度爆破，更加简单易用
MapReduce服务上线以来，我们在用户的意见和建议中不断地进行改进和优化。
快速构建基于 AWS Glue 的抽取跨区域 MySQL 8 的数据管道
在这个互联网高速发展，信息爆炸的时代，数据对于业务决策是至关重要的一环。
23｜数据库应用（三）：项目数据库配置实战
为了让你加深理解，今后轻松应用数据库来完成项目的数据管理和业务处理，接下来的两节课我会带你完成数据库配置，并结合项目综合应用数据库。
2023-06-14
教你如何在 Spark Scala/Java 应用中调用 Python 脚本
摘要：本文将介绍如何在Spark scala 程序中调用 Python 脚本，Spark java程序调用的过程也大体相同。
2021-11-08
发现更多内容
推荐阅读
电子书
大厂实战PPT下载
换一换 
2024：数据价值化技术趋势展望
史凯 | 《精益数据方法论》作者、价值驱动的精益数据治理体系创始人
立即下载
大模型+数据资产变现，RAG 驱动企业智能化实践案例
黄佳 | 新加坡科研局 Lead Research Engineer
立即下载
用 AI 重构 —— 大模型与低代码的“合体变身”
潘征 | 百度 百度资深研发工程师、爱速搭 AIGC+低代码实施方案负责人
立即下载
评论 1 条评论