编辑,修复适合composer-1.8.3和airflow-1.10.3的问题
在 Airflow 1.10.3 中,无法在外部创建集群配置。但是我们可以继承集群创建操作符并覆盖配置创建。这还允许我们设置可选组件,这是该 Airflow 版本中缺少的参数。
class CustomDataprocClusterCreateOperator(DataprocClusterCreateOperator):
def __init__(self, *args, **kwargs):
super(CustomDataprocClusterCreateOperator, self).__init__(*args, **kwargs)
def _build_cluster_data(self):
cluster_data = super(CustomDataprocClusterCreateOperator, self)._build_cluster_data()
cluster_data['config']['endpointConfig'] = {
'enableHttpPortAccess': True
}
cluster_data['config']['softwareConfig']['optionalComponents'] = [ 'JUPYTER', 'ANACONDA' ]
return cluster_data
#Start DataProc Cluster
dataproc_cluster = CustomDataprocClusterCreateOperator(
task_id='create-' + CLUSTER_NAME,
cluster_name=CLUSTER_NAME,
project_id=PROJECT_ID,
num_workers=3,
num_masters=1,
master_machine_type='n1-standard-2',
worker_machine_type='n1-standard-2',
master_disk_size=100,
worker_disk_size=100,
storage_bucket='test-dataproc-jupyter',
region='europe-west4',
zone='europe-west4-a',
auto_delete_ttl=21600,
dag=dag
)
原始答案,适用于 Airflow 1.10.7
虽然不是最佳选择,但您可以自己创建集群数据结构,而不是让 Airflow 的 ClusterGenerator 来完成此操作。它应该适用于最新版本(1.10.7)
cluster = {
'clusterName': CLUSTER_NAME,
'config': {
'gceClusterConfig': {
'zoneUri': 'europe-west4-a'
},
'masterConfig': {
'numInstances': 1,
'machineTypeUri': 'n1-standard-2',
'diskConfig': {
'bootDiskSizeGb': 100
},
},
'workerConfig': {
'numInstances': 3,
'machineTypeUri': 'n1-standard-2',
'diskConfig': {
'bootDiskSizeGb': 100
},
},
'softwareConfig': {
'optionalComponents': [
'ANACONDA',
'JUPYTER'
]
},
'lifestyleConfig': {
'autoDeleteTtl': 21600
},
'endpointConfig': {
'enableHttpPortAccess': True
}
},
'projectId': PROJECT_ID
}
#Start DataProc Cluster
dataproc_cluster = DataprocClusterCreateOperator(
task_id='create-' + CLUSTER_NAME,
project_id=PROJECT_ID,
num_workers=3,
region='europe-west4',
zone='europe-west4-a',
cluster = cluster,
dag=DAG
)
如果您使用其他 Airflow 版本,请注明。
您还可以为我打开的错误投票:AIRFLOW-6432