本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# 使用 DAG 在 CloudWatch 中编写自定义指标
您可以使用以下代码示例编写有向无环图（DAG），该图运行 `PythonOperator` 以检索 Amazon MWAA 环境的操作系统级指标。DAG 随后将数据作为自定义指标发布到 Amazon CloudWatch。
自定义操作系统级指标可让您进一步了解环境工作线程如何使用虚拟内存和 CPU 等资源。您可以使用此信息来选择最适合您的工作负载的[环境类](environment-class.md)。
**Topics**
+ [版本](#samples-custom-metrics-version)
+ [先决条件](#samples-custom-metrics-prereqs)
+ [权限](#samples-custom-metrics-permissions)
+ [依赖项](#samples-custom-metrics-dependencies)
+ [代码示例](#samples-custom-metrics-code)
## 版本
您可以在 [Python 3.10](https://peps.python.org/pep-0619/) 中将本页上的代码示例与 **Apache Airflow v2** 一起使用，在 [Python 3.11](https://peps.python.org/pep-0664/) 中与 **Apache Airflow v3** 一起使用。
## 先决条件
要使用本页上的代码示例，您需要以下内容：
+ [Amazon MWAA 环境](get-started.md)。
## 权限
无需其他权限即可使用本页上的代码示例。
## 依赖项
+ 无需其他依赖项即可使用本页上的代码示例。
## 代码示例
1. 在命令提示符下，导航到存储 DAG 代码的文件夹。例如：
```
cd dags
```
1. 复制以下代码示例的内容并本地另存为 `dag-custom-metrics.py`。用环境名称替换 `MWAA-ENV-NAME`。
```
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.utils.dates import days_ago
from datetime import datetime
import os,json,boto3,psutil,socket
def publish_metric(client,name,value,cat,unit='None'):
environment_name = os.getenv("MWAA_ENV_NAME")
value_number=float(value)
hostname = socket.gethostname()
ip_address = socket.gethostbyname(hostname)
print('writing value',value_number,'to metric',name)
response = client.put_metric_data(
Namespace='MWAA-Custom',
MetricData=[
{
'MetricName': name,
'Dimensions': [
{
'Name': 'Environment',
'Value': environment_name
},
{
'Name': 'Category',
'Value': cat
},
{
'Name': 'Host',
'Value': ip_address
},
],
'Timestamp': datetime.now(),
'Value': value_number,
'Unit': unit
},
]
)
print(response)
return response
def python_fn(**kwargs):
client = boto3.client('cloudwatch')
cpu_stats = psutil.cpu_stats()
print('cpu_stats', cpu_stats)
virtual = psutil.virtual_memory()
cpu_times_percent = psutil.cpu_times_percent(interval=0)
publish_metric(client=client, name='virtual_memory_total', cat='virtual_memory', value=virtual.total, unit='Bytes')
publish_metric(client=client, name='virtual_memory_available', cat='virtual_memory', value=virtual.available, unit='Bytes')
publish_metric(client=client, name='virtual_memory_used', cat='virtual_memory', value=virtual.used, unit='Bytes')
publish_metric(client=client, name='virtual_memory_free', cat='virtual_memory', value=virtual.free, unit='Bytes')
publish_metric(client=client, name='virtual_memory_active', cat='virtual_memory', value=virtual.active, unit='Bytes')
publish_metric(client=client, name='virtual_memory_inactive', cat='virtual_memory', value=virtual.inactive, unit='Bytes')
publish_metric(client=client, name='virtual_memory_percent', cat='virtual_memory', value=virtual.percent, unit='Percent')
publish_metric(client=client, name='cpu_times_percent_user', cat='cpu_times_percent', value=cpu_times_percent.user, unit='Percent')
publish_metric(client=client, name='cpu_times_percent_system', cat='cpu_times_percent', value=cpu_times_percent.system, unit='Percent')
publish_metric(client=client, name='cpu_times_percent_idle', cat='cpu_times_percent', value=cpu_times_percent.idle, unit='Percent')
return "OK"
with DAG(dag_id=os.path.basename(__file__).replace(".py", ""), schedule_interval='*/5 * * * *', catchup=False, start_date=days_ago(1)) as dag:
t = PythonOperator(task_id="memory_test", python_callable=python_fn, provide_context=True)
```
1. 运行以下 AWS CLI 命令将 DAG 复制到环境的存储桶，然后使用 Apache Airflow UI 触发 DAG。
```
aws s3 cp your-dag.py s3://your-environment-bucket/dags/
```
1. 如果 DAG 成功运行，您会在 Apache Airflow 日志中收到类似以下的内容：
```
[2022-08-16, 10:54:46 UTC] {{logging_mixin.py:109}} INFO - cpu_stats scpustats(ctx_switches=3253992384, interrupts=1964237163, soft_interrupts=492328209, syscalls=0)
[2022-08-16, 10:54:46 UTC] {{logging_mixin.py:109}} INFO - writing value 16024199168.0 to metric virtual_memory_total
[2022-08-16, 10:54:46 UTC] {{logging_mixin.py:109}} INFO - {'ResponseMetadata': {'RequestId': 'fad289ac-aa51-46a9-8b18-24e4e4063f4d', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': 'fad289ac-aa51-46a9-8b18-24e4e4063f4d', 'content-type': 'text/xml', 'content-length': '212', 'date': 'Tue, 16 Aug 2022 17:54:45 GMT'}, 'RetryAttempts': 0}}
[2022-08-16, 10:54:46 UTC] {{logging_mixin.py:109}} INFO - writing value 14356287488.0 to metric virtual_memory_available
[2022-08-16, 10:54:46 UTC] {{logging_mixin.py:109}} INFO - {'ResponseMetadata': {'RequestId': '6ef60085-07ab-4865-8abf-dc94f90cab46', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': '6ef60085-07ab-4865-8abf-dc94f90cab46', 'content-type': 'text/xml', 'content-length': '212', 'date': 'Tue, 16 Aug 2022 17:54:45 GMT'}, 'RetryAttempts': 0}}
[2022-08-16, 10:54:46 UTC] {{logging_mixin.py:109}} INFO - writing value 1342296064.0 to metric virtual_memory_used
[2022-08-16, 10:54:46 UTC] {{logging_mixin.py:109}} INFO - {'ResponseMetadata': {'RequestId': 'd5331438-5d3c-4df2-bc42-52dcf8d60a00', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': 'd5331438-5d3c-4df2-bc42-52dcf8d60a00', 'content-type': 'text/xml', 'content-length': '212', 'date': 'Tue, 16 Aug 2022 17:54:45 GMT'}, 'RetryAttempts': 0}}
...
[2022-08-16, 10:54:46 UTC] {{python.py:152}} INFO - Done. Returned value was: OK
[2022-08-16, 10:54:46 UTC] {{taskinstance.py:1280}} INFO - Marking task as SUCCESS. dag_id=dag-custom-metrics, task_id=memory_test, execution_date=20220816T175444, start_date=20220816T175445, end_date=20220816T175446
[2022-08-16, 10:54:46 UTC] {{local_task_job.py:154}} INFO - Task exited with return code 0
```