本文首发于：行者AI
AWS EMR是一个计算集群。能够通过ta创立自定义配置的虚拟机，并主动装置所需计算框架（Spark，Hadoop，Hive等），以便用来进行大数据计算。
1. 我的项目背景
公司目前有一个我的项目，通过爬虫收集数据，离线计算失去用户画像，并将最终后果写入rds，通过api向外展现数据。
2. 架构演进
2.1 技术栈
- 计算框架 Spark
- 调度框架 Airflow
- 数据存储 Hadoop，Mysql
- 数仓工具 Hive，Presto
- 辅助工具 Zepplin
- 脚本语言 Java，Scala，Python
2.2 第一版
环境
咱们在某云厂商开了6台虚构器（4核8G），spark on yarn模式运行，其中1台作为主节点，运行hadoop主节点和airflow调度程序，其余作为数据节点。
计算过程
- 通过Spark Streaming将数据落地到Hadoop
- Airflow定时向主节点通过Spark-submit形式提交命令
- Spark计算后将最终后果写入Mysql
- 平时开发人员能够在Zepplin进行查问
成果
计算流程能够失常进行
思考
通过一段时间的察看剖析，咱们发现
- 大部分计算工作都能在较短时间内实现
- 机器每天闲置工夫很长
- 业务没有很高的实时性要求
- 高配置虚构器老本很高
论断
基于现状，咱们心愿能有个即开即用的零碎，就像电脑一样，要用就关上，用完就敞开。通过调研，最终抉择了AWS的EMR。
2.3 第二版
环境
在将零碎迁徙到AWS EMR之后，在AWS上开了一台虚构器（1核2G）运行Airflow和Kinesis
这台虚构器须要始终运行，但Airflow自身不须要高配置
计算过程
- 通过Kinesis将数据落到S3
Airflow定时发动工作
- 发动创立EMR申请
可自定义机器配置，要装置的计算框架，也可笼罩框架配置。可通过Python脚本检测集群是否创立胜利
- 提交计算工作
- 发动创立EMR申请
- 敞开集群
成果
计算流程能够失常进行，但不须要长开机器了，只须要一台低配来触发定时工作即可
思考
通过一段时间的察看
- EMR费用比起虚构器，的确便宜很多
- 能够通过console台查看集群状态，管制集群开关
- 不不便的中央，平时要查看Hadoop的数据，须要本人写脚本拉取，不能应用辅助工具了
::: hljs-center
Talk is cheap, show me the code
:::
筹备工作
- 注册AWS账号，登录
- 开明EMR，S3
开明S3的目标是为了长久化数据，因为EMR集群自身不带额定硬盘，须要内部介质贮存
- 开明AWS内网可拜访的Mysql
如果不必Hive，可跳过这一步，同理，须要内部介质贮存Hive的数据结构
- 筹备创立EMR集群的脚本
这里有个坑，开始咱们应用的AWS SDK来做这件事，但无奈自定义计算框架配置（应该是BUG），最后咱们通过批改SDK源码解决了这个问题，但起初发现根本没用到SDK其余性能时，咱们将这部分代码提成了独自的文件，因为应用了Airflow进行调度，所以决定用了Python
- 编写Spark工作，打包上传至S3
EMR LIB
# coding: UTF-8import boto3, json, requests, requestsfrom datetime import datetimedef get_region(): # 这个地址不必改 r = requests.get("http://169.254.169.254/latest/dynamic/instance-identity/document") response_json = r.json() return response_json.get('region')def client(region_name): global emr emr = boto3.client('emr', region_name=region_name)# 创立EMRdef create_cluster(name): param = { # 批改须要的框架 "Applications":[{ "Name":"Hadoop" },{ "Name":"Hive" },{ "Name":"Spark" }], # 这里的名字会显示到控制台 "Name":name, "ServiceRole":"EMR_DefaultRole", "Tags":[], "ReleaseLabel":"emr-5.26.0", "Instances":{ "TerminationProtected":False, "EmrManagedMasterSecurityGroup":"sg-0085fba9c3a6818f5", "InstanceGroups":[{ "InstanceCount":1, "Name":"主实例组 - 1", "InstanceRole":"MASTER", "EbsConfiguration":{ "EbsBlockDeviceConfigs":[{ "VolumeSpecification":{ "SizeInGB":32, "VolumeType":"gp2" }, "VolumesPerInstance":1 }] }, # 批改须要的硬件配置 "InstanceType":"m4.large", "Market":"ON_DEMAND", "Configurations":[{ # 批改Hive的meta源 "Classification":"hive-site", "Properties":{ "javax.jdo.option.ConnectionURL":"jdbc:mysql://host:port/db?useUnicode=true&characterEncoding=UTF-8", "javax.jdo.option.ConnectionDriverName":"org.mariadb.jdbc.Driver", "javax.jdo.option.ConnectionUserName":"user", "javax.jdo.option.ConnectionPassword":"pwd" } },{ "Classification":"yarn-env", "Properties":{}, "Configurations":[{ "Classification":"export", "Properties":{ "AWS_REGION":"cn-northwest-1", "S3_ENDPOINT":"s3.cn-northwest-1.amazonaws.com.cn", "S3_USE_HTTPS":"0", "S3_VERIFY_SSL":"0" } }] }] },{ "InstanceRole":"CORE", "InstanceCount":1, "Name":"外围实例组 - 2", "Market":"ON_DEMAND", # 批改须要的硬件配置 "InstanceType":"r5d.2xlarge", "Configurations":[{ "Classification":"hive-site", "Properties":{ "javax.jdo.option.ConnectionURL":"jdbc:mysql://host:port/db?useUnicode=true&characterEncoding=UTF-8", "javax.jdo.option.ConnectionDriverName":"org.mariadb.jdbc.Driver", "javax.jdo.option.ConnectionUserName":"user", "javax.jdo.option.ConnectionPassword":"pwd" } },{ "Classification":"yarn-env", "Properties":{}, "Configurations":[{ "Classification":"export", "Properties":{ "AWS_REGION":"cn-northwest-1", "S3_ENDPOINT":"s3.cn-northwest-1.amazonaws.com.cn", "S3_USE_HTTPS":"0", "S3_VERIFY_SSL":"0" } }] }] },{ # 批改须要的工作节点数 "InstanceCount":4, "Name":"工作实例组 - 4", "InstanceRole":"TASK", "EbsConfiguration":{ "EbsBlockDeviceConfigs":[{ "VolumeSpecification":{ "SizeInGB":32, "VolumeType":"gp2" }, "VolumesPerInstance":4 }] }, # 批改须要的硬件配置 "InstanceType":"r5d.2xlarge", "Market":"ON_DEMAND", "Configurations":[{ "Classification":"hive-site", "Properties":{ "javax.jdo.option.ConnectionURL":"jdbc:mysql://host:port/db?useUnicode=true&characterEncoding=UTF-8", "javax.jdo.option.ConnectionDriverName":"org.mariadb.jdbc.Driver", "javax.jdo.option.ConnectionUserName":"user", "javax.jdo.option.ConnectionPassword":"pwd" } },{ "Classification":"yarn-env", "Properties":{}, "Configurations":[{ "Classification":"export", "Properties":{ "AWS_REGION":"cn-northwest-1", "S3_ENDPOINT":"s3.cn-northwest-1.amazonaws.com.cn", "S3_USE_HTTPS":"0", "S3_VERIFY_SSL":"0" } }] }] }], "KeepJobFlowAliveWhenNoSteps":True, "Ec2SubnetId":"subnet-027bff297ea95039b", "Ec2KeyName":"hifive.airflow", "EmrManagedSlaveSecurityGroup":"sg-05a0e076ee7babb9e" }, "JobFlowRole":"EMR_EC2_DefaultRole", "Steps":[{ "HadoopJarStep":{ "Args":["state-pusher-script"], "Jar":"command-runner.jar" }, "Name":"Setup Hadoop Debugging" }], "ScaleDownBehavior":"TERMINATE_AT_TASK_COMPLETION", "VisibleToAllUsers":True, "EbsRootVolumeSize":10, "LogUri":"s3n://aws-logs-550775287661-cn-northwest-1/elasticmapreduce/", "AutoScalingRole":"EMR_AutoScaling_DefaultRole" } cluster_response = emr.run_job_flow(**param) return cluster_response['JobFlowId']# 获取EMR拜访入口def get_cluster_dns(cluster_id): response = emr.describe_cluster(ClusterId=cluster_id) return response['Cluster']['MasterPublicDnsName']# 期待集群创立实现def wait_for_cluster_creation(cluster_id): emr.get_waiter('cluster_running').wait(ClusterId=cluster_id)# 敞开EMRdef terminate_cluster(cluster_id): emr.terminate_job_flows(JobFlowIds=[cluster_id])调用测试
# 创立6台机器的集群（1 master，1 core，4 worker）cluster_id = create_cluster("biz_daily_2020_10_09")# 阻塞直到创立胜利wait_for_cluster_creation(cluster_id)# dns相当于虚拟机的ssh地址，每次都不同# ssh登录这个地址能够提交spark命令了，这里应用Airflow的SSHOperator模仿登录并提交命令cluster_dns = get_cluster_dns(cluster_id)# 敞开集群terminate_cluster(cluster_id)3. 其余坑
- Airflow 1.9.0的工夫模板{{ ds }}生成的是格林尼治工夫，要改为我国工夫，需手动加8小时，不晓得新版本是否反对本地工夫。
- ssh登录dns用户名hadoop，这个用户是AWS生成的，仿佛无奈批改。