기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

# 오토스케일러 파라미터 자동 조정
<a name="jobruns-flink-autoscaler-parameter-autotuning"></a>

이 섹션에서는 다양한 Amazon EMR 버전에 대한 자동 조정 동작을 설명합니다. 또한 다양한 오토 스케일링 구성도 자세히 설명합니다.

**참고**  
Amazon EMR 7.2.0 이상은 오픈 소스 구성 `job.autoscaler.restart.time-tracking.enabled`를 사용하여 **재조정 시간 예측**을 활성화합니다. 재조정 시간 예측은 Amazon EMR 자동 조정과 동일한 기능을 제공하므로, 재시작 시간에 경험적 값을 수동으로 할당하지 않아도 됩니다.  
Amazon EMR 7.1.0 이하를 사용하는 경우에도 Amazon EMR 자동 조정을 계속 사용할 수 있습니다.

------
#### [ 7.2.0 and higher ]

Amazon EMR 7.2.0 이상은 오토 스케일링 결정을 적용하는 데 필요한 실제 재시작 시간을 측정합니다. 릴리스 7.1.0 이하에서는 `job.autoscaler.restart.time` 구성을 사용하여 예상 최대 재시작 시간을 수동으로 구성해야 했습니다. `job.autoscaler.restart.time-tracking.enabled` 구성을 사용하면 첫 번째 조정에 대한 재시작 시간만 입력하면 됩니다. 이후 작업자는 실제 재시작 시간을 기록하고 후속 조정에 사용합니다.

이 추적을 활성화하려면 다음 명령을 사용합니다.

```
job.autoscaler.restart.time-tracking.enabled: true
```

다음은 재조정 시간 예측을 위한 관련 구성입니다.


| 구성 | 필수 | 기본값 | 설명 | 
| --- | --- | --- | --- | 
| job.autoscaler.restart.time-tracking.enabled | 아니요 | False | Flink 오토스케일러가 시간 경과에 따라 구성을 자동으로 조정하여 조정 결정을 최적화해야 하는지 여부를 나타냅니다. 오토스케일러는 오토스케일러 파라미터 restart.time만 자동 조정할 수 있습니다. | 
| job.autoscaler.restart.time | 아니요 | 5분 | 연산자가 이전 조정에서 실제 재시작 시간을 확인할 수 있을 때까지 Amazon EMR on EKS에서 사용하는 예상 재시작 시간. | 
| job.autoscaler.restart.time-tracking.limit | 아니요 | 15분 | job.autoscaler.restart.time-tracking.enabled가 true로 설정된 경우 관찰된 최대 재시작 시간. | 

다음은 재조정 시간 예측을 시도하는 데 사용할 수 있는 배포 사양의 예입니다.

```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: autoscaling-example
spec:
  flinkVersion: v1_18
  flinkConfiguration:

    # Autoscaler parameters
    job.autoscaler.enabled: "true"
    job.autoscaler.scaling.enabled: "true"
    job.autoscaler.stabilization.interval: "5s"
    job.autoscaler.metrics.window: "1m"
    
    job.autoscaler.restart.time-tracking.enabled: "true"
    job.autoscaler.restart.time: "2m"
    job.autoscaler.restart.time-tracking.limit: "10m"
    
    jobmanager.scheduler: adaptive
    taskmanager.numberOfTaskSlots: "1"
    pipeline.max-parallelism: "12"

  executionRoleArn: {{<JOB ARN>}}
  emrReleaseLabel: emr-7.13.0-flink-latest
  jobManager:
    highAvailabilityEnabled: false
    storageDir: s3://{{<s3_bucket>}}/flink/autoscaling/ha/
    replicas: 1
    resource:
      memory: "1024m"
      cpu: 0.5
  taskManager:
    resource:
      memory: "1024m"
      cpu: 0.5
  job:
    jarURI: s3://{{<s3_bucket>}}/some-job-with-back-pressure
    parallelism: 1
    upgradeMode: stateless
```

역압을 시뮬레이션하려면 다음 배포 사양을 사용합니다.

```
job:
    jarURI: s3://{{<s3_bucket>}}/pyflink-script.py
    entryClass: "org.apache.flink.client.python.PythonDriver"
    args: ["-py", "/opt/flink/usrlib/pyflink-script.py"]
    parallelism: 1
    upgradeMode: stateless
```

다음 Python 스크립트를 S3 버킷에 업로드합니다.

```
import logging
import sys
import time
import random

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

TABLE_NAME="orders"
QUERY=f"""
CREATE TABLE {TABLE_NAME} (
  id INT,
  order_time AS CURRENT_TIMESTAMP,
  WATERMARK FOR order_time AS order_time - INTERVAL '5' SECONDS
)
WITH (
  'connector' = 'datagen',
  'rows-per-second'='10',
  'fields.id.kind'='random',
  'fields.id.min'='1',
  'fields.id.max'='100'
);
"""

def create_backpressure(i):
    time.sleep(2)
    return i

def autoscaling_demo():
    env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(env)
    t_env.execute_sql(QUERY)
    res_table = t_env.from_path(TABLE_NAME)

    stream =  t_env.to_data_stream(res_table) \
      .shuffle().map(lambda x: create_backpressure(x))\
      .print()
    env.execute("Autoscaling demo")

if __name__ == '__main__':
    logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")
    autoscaling_demo()
```

재조정 시간 예측이 작동하는지 확인하려면 Flink 연산자의 `DEBUG` 수준 로깅이 활성화되어 있는지 확인합니다. 아래 예제에서는 헬름 차트 파일(`values.yaml`)을 업데이트하는 방법을 보여줍니다. 그런 다음, 업데이트된 헬름 차트를 다시 설치하고 Flink 작업을 다시 실행합니다.

```
log4j-operator.properties: |+
  # Flink Operator Logging Overrides
  rootLogger.level = DEBUG
```

리더 포드의 이름을 가져옵니다.

```
ip=$(kubectl get configmap -n $NAMESPACE {{<job-name>}}-cluster-config-map -o json | jq -r ".data[\"org.apache.flink.k8s.leader.restserver\"]" | awk -F: '{print $2}' | awk -F '/' '{print $3}')

kubectl get pods -n $NAMESPACE -o json | jq -r ".items[] | select(.status.podIP == \"$ip\") | .metadata.name"
```

다음 명령을 실행하여 지표 평가에 사용된 실제 재시작 시간을 가져옵니다.

```
kubectl logs {{<FLINK-OPERATOR-POD-NAME>}} -c flink-kubernetes-operator -n {{<OPERATOR-NAMESPACE>}} -f | grep "Restart time used in scaling summary computation"
```

다음과 비슷한 로그가 표시됩니다. 첫 번째 조정에서만 ` job.autoscaler.restart.time`을 사용합니다. 후속 조정에서는 관찰된 재시작 시간을 사용합니다.

```
2024-05-16 17:17:32,590 o.a.f.a.ScalingExecutor        [DEBUG][default/autoscaler-example] Restart time used in scaling summary computation: PT2M
2024-05-16 17:19:03,787 o.a.f.a.ScalingExecutor        [DEBUG][default/autoscaler-example] Restart time used in scaling summary computation: PT14S
2024-05-16 17:19:18,976 o.a.f.a.ScalingExecutor        [DEBUG][default/autoscaler-example] Restart time used in scaling summary computation: PT14S
2024-05-16 17:20:50,283 o.a.f.a.ScalingExecutor        [DEBUG][default/autoscaler-example] Restart time used in scaling summary computation: PT14S
2024-05-16 17:22:21,691 o.a.f.a.ScalingExecutor        [DEBUG][default/autoscaler-example] Restart time used in scaling summary computation: PT14S
```

------
#### [ 7.0.0 and 7.1.0 ]

오픈 소스 기본 제공 Flink 오토스케일러는 다양한 지표를 사용하여 최적의 조정 결정을 내립니다. 그러나 계산에 사용하는 기본값은 대부분의 워크로드에 적용 가능한 값이지만, 주어진 작업에 적합하지 않을 수 있습니다. Flink 연산자의 Amazon EMR on EKS 버전에 추가된 자동 조정 기능은 캡처된 특정 지표에서 관찰된 과거 추세를 검토한 다음, 해당 작업에 맞게 조정된 가장 최적의 값을 계산하려고 시도합니다.


| 구성 | 필수 | 기본값 | 설명 | 
| --- | --- | --- | --- | 
| kubernetes.operator.job.autoscaler.autotune.enable | 아니요 | False | Flink 오토스케일러가 시간 경과에 따라 구성을 자동으로 조정하여 오토스케일러 조정 결정을 최적화해야 하는지 여부를 나타냅니다. 현재 오토스케일러는 오토스케일러 파라미터 restart.time만 자동 조정할 수 있습니다. | 
| kubernetes.operator.job.autoscaler.autotune.metrics.history.max.count | 아니요 | 3 | 오토스케일러가 Amazon EMR on EKS 지표 구성 맵에 보관하는 Amazon EMR on EKS의 기록 지표 수를 나타냅니다. | 
| kubernetes.operator.job.autoscaler.autotune.metrics.restart.count | 아니요 | 3 | 주어진 작업의 평균 재시작 시간 계산을 시작하기 전에 오토스케일러가 수행하는 재시작 횟수를 나타냅니다. | 

자동 조정을 활성화하려면 다음을 완료해야 합니다.
+ `kubernetes.operator.job.autoscaler.autotune.enable:`를 `true`로 설정합니다.
+ `metrics.job.status.enable:`를 `TOTAL_TIME`로 설정합니다.
+ 자동 조정을 활성화하기 위해 [Flink 애플리케이션에 대해 오토스케이러 사용](https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/jobruns-flink-autoscaler.html) 설정을 따랐습니다.

다음은 자동 조정을 시도하는 데 사용할 수 있는 배포 사양에 대한 예제입니다.

```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: autoscaling-example
spec:
  flinkVersion: v1_18
  flinkConfiguration:

    # Autotuning parameters
    kubernetes.operator.job.autoscaler.autotune.enable: "true"
    kubernetes.operator.job.autoscaler.autotune.metrics.history.max.count: "2"
    kubernetes.operator.job.autoscaler.autotune.metrics.restart.count: "1"
    metrics.job.status.enable: TOTAL_TIME

    # Autoscaler parameters
    kubernetes.operator.job.autoscaler.enabled: "true"
    kubernetes.operator.job.autoscaler.scaling.enabled: "true"
    kubernetes.operator.job.autoscaler.stabilization.interval: "5s"
    kubernetes.operator.job.autoscaler.metrics.window: "1m"

    jobmanager.scheduler: adaptive

    taskmanager.numberOfTaskSlots: "1"
    state.savepoints.dir: s3://<S3_bucket>/autoscaling/savepoint/
    state.checkpoints.dir: s3://<S3_bucket>/flink/autoscaling/checkpoint/
    pipeline.max-parallelism: "4"

  executionRoleArn: <JOB ARN>
  emrReleaseLabel: emr-6.14.0-flink-latest
  jobManager:
    highAvailabilityEnabled: true
    storageDir: s3://<S3_bucket>/flink/autoscaling/ha/
    replicas: 1
    resource:
      memory: "1024m"
      cpu: 0.5
  taskManager:
    resource:
      memory: "1024m"
      cpu: 0.5
  job:
    jarURI: s3://<S3_bucket>/some-job-with-back-pressure
    parallelism: 1
    upgradeMode: last-state
```

역압을 시뮬레이션하려면 다음 배포 사양을 사용합니다.

```
  job:
    jarURI: s3://<S3_bucket>/pyflink-script.py
    entryClass: "org.apache.flink.client.python.PythonDriver"
    args: ["-py", "/opt/flink/usrlib/pyflink-script.py"]
    parallelism: 1
    upgradeMode: last-state
```

다음 Python 스크립트를 S3 버킷에 업로드합니다.

```
import logging
import sys
import time
import random

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

TABLE_NAME="orders"
QUERY=f"""
CREATE TABLE {TABLE_NAME} (
  id INT,
  order_time AS CURRENT_TIMESTAMP,
  WATERMARK FOR order_time AS order_time - INTERVAL '5' SECONDS
)
WITH (
  'connector' = 'datagen',
  'rows-per-second'='10',
  'fields.id.kind'='random',
  'fields.id.min'='1',
  'fields.id.max'='100'
);
"""

def create_backpressure(i):
    time.sleep(2)
    return i

def autoscaling_demo():
    env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(env)
    t_env.execute_sql(QUERY)
    res_table = t_env.from_path(TABLE_NAME)

    stream =  t_env.to_data_stream(res_table) \
      .shuffle().map(lambda x: create_backpressure(x))\
      .print()
    env.execute("Autoscaling demo")

if __name__ == '__main__':
    logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")
    autoscaling_demo()
```

자동 튜너가 작동하는지 확인하려면 다음 명령을 사용합니다. Flink 연산자에 대한 자체 리더 포드 정보를 사용해야 합니다.

먼저 리더 포드의 이름을 가져옵니다.

```
ip=$(kubectl get configmap -n $NAMESPACE <job-name>-cluster-config-map -o json | jq -r ".data[\"org.apache.flink.k8s.leader.restserver\"]" | awk -F: '{print $2}' | awk -F '/' '{print $3}')

kubectl get pods -n $NAMESPACE -o json | jq -r ".items[] | select(.status.podIP == \"$ip\") | .metadata.name"
```

리더 포드의 이름을 지정한 후 다음 명령을 실행할 수 있습니다.

```
kubectl logs -n $NAMESPACE  -c flink-kubernetes-operator --follow <{{YOUR-FLINK-OPERATOR-POD-NAME}}>  | grep -E 'EmrEks|autotun|calculating|restart|autoscaler'
```

다음과 비슷한 로그가 표시됩니다.

```
[m[33m2023-09-13 20:10:35,941[m [36mc.a.c.f.k.o.a.EmrEksMetricsAutotuner[m [36m[DEBUG][flink/autoscaling-example] Using the latest Emr Eks Metric for calculating restart.time for autotuning: EmrEksMetrics(restartMetric=RestartMetric(restartingTime=65, numRestarts=1))

[m[33m2023-09-13 20:10:35,941[m [36mc.a.c.f.k.o.a.EmrEksMetricsAutotuner[m [32m[INFO ][flink/autoscaling-example] Calculated average restart.time metric via autotuning to be: PT0.065S
```

------