Reformat the whole AWS documentation (#23810)

airflow/providers/amazon/aws/example_dags/example_athena.py

@@ -84,21 +84,21 @@ def read_results_from_s3(query_execution_id):
         output_location=f's3://{S3_BUCKET}/{S3_KEY}',
     )
 
-    # [START howto_athena_operator]
+    # [START howto_operator_athena]
     read_table = AthenaOperator(
         task_id='read_table',
         query=QUERY_READ_TABLE,
         database=ATHENA_DATABASE,
         output_location=f's3://{S3_BUCKET}/{S3_KEY}',
     )
-    # [END howto_athena_operator]
+    # [END howto_operator_athena]
 
-    # [START howto_athena_sensor]
+    # [START howto_sensor_athena]
     await_query = AthenaSensor(
         task_id='await_query',
         query_execution_id=read_table.output,
     )
-    # [END howto_athena_sensor]
+    # [END howto_sensor_athena]
 
     drop_table = AthenaOperator(
         task_id='drop_table',

airflow/providers/amazon/aws/example_dags/example_dms.py

@@ -319,12 +319,12 @@ def clean_up():
     # [END howto_operator_dms_stop_task]
 
     # TaskCompletedSensor actually waits until task reaches the "Stopped" state, so it will work here.
-    # [START howto_operator_dms_task_completed_sensor]
+    # [START howto_sensor_dms_task_completed]
     await_task_stop = DmsTaskCompletedSensor(
         task_id='await_task_stop',
         replication_task_arn=create_task.output,
     )
-    # [END howto_operator_dms_task_completed_sensor]
+    # [END howto_sensor_dms_task_completed]
 
     # [START howto_operator_dms_delete_task]
     delete_task = DmsDeleteTaskOperator(

airflow/providers/amazon/aws/example_dags/example_dynamodb_to_s3.py

@@ -18,6 +18,7 @@
 from os import environ
 
 from airflow import DAG
+from airflow.models.baseoperator import chain
 from airflow.providers.amazon.aws.transfers.dynamodb_to_s3 import DynamoDBToS3Operator
 
 TABLE_NAME = environ.get('DYNAMO_TABLE_NAME', 'ExistingDynamoDbTableName')
@@ -31,7 +32,6 @@
     tags=['example'],
     catchup=False,
 ) as dag:
-
     # [START howto_transfer_dynamodb_to_s3]
     backup_db = DynamoDBToS3Operator(
         task_id='backup_db',
@@ -41,3 +41,32 @@
         file_size=1000,
     )
     # [END howto_transfer_dynamodb_to_s3]
+
+    # [START howto_transfer_dynamodb_to_s3_segmented]
+    # Segmenting allows the transfer to be parallelized into {segment} number of parallel tasks.
+    backup_db_segment_1 = DynamoDBToS3Operator(
+        task_id='backup-1',
+        dynamodb_table_name=TABLE_NAME,
+        s3_bucket_name=BUCKET_NAME,
+        # Max output file size in bytes.  If the Table is too large, multiple files will be created.
+        file_size=1000,
+        dynamodb_scan_kwargs={
+            "TotalSegments": 2,
+            "Segment": 0,
+        },
+    )
+
+    backup_db_segment_2 = DynamoDBToS3Operator(
+        task_id="backup-2",
+        dynamodb_table_name=TABLE_NAME,
+        s3_bucket_name=BUCKET_NAME,
+        # Max output file size in bytes.  If the Table is too large, multiple files will be created.
+        file_size=1000,
+        dynamodb_scan_kwargs={
+            "TotalSegments": 2,
+            "Segment": 1,
+        },
+    )
+    # [END howto_transfer_dynamodb_to_s3_segmented]
+
+    chain(backup_db, [backup_db_segment_1, backup_db_segment_2])

airflow/providers/amazon/aws/example_dags/example_glacier_to_gcs.py

@@ -32,20 +32,20 @@
     start_date=datetime(2021, 1, 1),  # Override to match your needs
     catchup=False,
 ) as dag:
-    # [START howto_glacier_create_job_operator]
+    # [START howto_operator_glacier_create_job]
     create_glacier_job = GlacierCreateJobOperator(task_id="create_glacier_job", vault_name=VAULT_NAME)
     JOB_ID = '{{ task_instance.xcom_pull("create_glacier_job")["jobId"] }}'
-    # [END howto_glacier_create_job_operator]
+    # [END howto_operator_glacier_create_job]
 
-    # [START howto_glacier_job_operation_sensor]
+    # [START howto_sensor_glacier_job_operation]
     wait_for_operation_complete = GlacierJobOperationSensor(
         vault_name=VAULT_NAME,
         job_id=JOB_ID,
         task_id="wait_for_operation_complete",
     )
-    # [END howto_glacier_job_operation_sensor]
+    # [END howto_sensor_glacier_job_operation]
 
-    # [START howto_glacier_transfer_data_to_gcs]
+    # [START howto_transfer_glacier_to_gcs]
     transfer_archive_to_gcs = GlacierToGCSOperator(
         task_id="transfer_archive_to_gcs",
         vault_name=VAULT_NAME,
@@ -57,6 +57,6 @@
         # then whole file will be downloaded
         chunk_size=1024,
     )
-    # [END howto_glacier_transfer_data_to_gcs]
+    # [END howto_transfer_glacier_to_gcs]
 
     create_glacier_job >> wait_for_operation_complete >> transfer_archive_to_gcs

airflow/providers/amazon/aws/example_dags/example_google_api_sheets_to_s3.py

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 """
-This is a basic example dag for using `GoogleApiToS3Transfer` to retrieve Google Sheets data
+This is a basic example dag for using `GoogleApiToS3Operator` to retrieve Google Sheets data
 You need to set all env variables to request the data.
 """
 

airflow/providers/amazon/aws/example_dags/example_google_api_youtube_to_s3.py

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 """
-This is a more advanced example dag for using `GoogleApiToS3Transfer` which uses xcom to pass data between
+This is a more advanced example dag for using `GoogleApiToS3Operator` which uses xcom to pass data between
 tasks to retrieve specific information about YouTube videos:
 
 First it searches for up to 50 videos (due to pagination) in a given time range

airflow/providers/amazon/aws/example_dags/example_lambda.py

@@ -36,10 +36,10 @@
     tags=['example'],
     catchup=False,
 ) as dag:
-    # [START howto_lambda_operator]
+    # [START howto_operator_lambda]
     invoke_lambda_function = AwsLambdaInvokeFunctionOperator(
         task_id='setup__invoke_lambda_function',
         function_name=LAMBDA_FUNCTION_NAME,
         payload=SAMPLE_EVENT,
     )
-    # [END howto_lambda_operator]
+    # [END howto_operator_lambda]

airflow/providers/amazon/aws/example_dags/example_redshift_data_execute_sql.py

@@ -57,7 +57,7 @@ def output_query_results(statement_id):
     catchup=False,
     tags=['example'],
 ) as dag:
-    # [START howto_redshift_data]
+    # [START howto_operator_redshift_data]
     task_query = RedshiftDataOperator(
         task_id='redshift_query',
         cluster_identifier=REDSHIFT_CLUSTER_IDENTIFIER,
@@ -67,6 +67,6 @@ def output_query_results(statement_id):
         poll_interval=POLL_INTERVAL,
         await_result=True,
     )
-    # [END howto_redshift_data]
+    # [END howto_operator_redshift_data]
 
     task_output = output_query_results(task_query.output)

airflow/providers/amazon/aws/example_dags/example_sagemaker.py

@@ -393,12 +393,12 @@ def cleanup():
     )
     # [END howto_operator_sagemaker_training]
 
-    # [START howto_operator_sagemaker_training_sensor]
+    # [START howto_sensor_sagemaker_training]
     await_training = SageMakerTrainingSensor(
         task_id='await_training',
         job_name=TRAINING_JOB_NAME,
     )
-    # [END howto_operator_sagemaker_training_sensor]
+    # [END howto_sensor_sagemaker_training]
 
     # [START howto_operator_sagemaker_model]
     create_model = SageMakerModelOperator(
@@ -418,12 +418,12 @@ def cleanup():
     )
     # [END howto_operator_sagemaker_tuning]
 
-    # [START howto_operator_sagemaker_tuning_sensor]
+    # [START howto_sensor_sagemaker_tuning]
     await_tune = SageMakerTuningSensor(
         task_id='await_tuning',
         job_name=TUNING_JOB_NAME,
     )
-    # [END howto_operator_sagemaker_tuning_sensor]
+    # [END howto_sensor_sagemaker_tuning]
 
     # [START howto_operator_sagemaker_transform]
     test_model = SageMakerTransformOperator(
@@ -435,12 +435,12 @@ def cleanup():
     )
     # [END howto_operator_sagemaker_transform]
 
-    # [START howto_operator_sagemaker_transform_sensor]
+    # [START howto_sensor_sagemaker_transform]
     await_transform = SageMakerTransformSensor(
         task_id='await_transform',
         job_name=TRANSFORM_JOB_NAME,
     )
-    # [END howto_operator_sagemaker_transform_sensor]
+    # [END howto_sensor_sagemaker_transform]
 
     # Trigger rule set to "all_done" so clean up will run regardless of success on other tasks.
     # [START howto_operator_sagemaker_delete_model]

airflow/providers/amazon/aws/example_dags/example_sagemaker_endpoint.py

@@ -202,13 +202,13 @@ def cleanup():
     )
     # [END howto_operator_sagemaker_endpoint]
 
-    # [START howto_operator_sagemaker_endpoint_sensor]
+    # [START howto_sensor_sagemaker_endpoint]
     await_endpoint = SageMakerEndpointSensor(
         task_id='await_endpoint',
         endpoint_name=ENDPOINT_NAME,
         do_xcom_push=False,
     )
-    # [END howto_operator_sagemaker_endpoint_sensor]
+    # [END howto_sensor_sagemaker_endpoint]
 
     # Trigger rule set to "all_done" so clean up will run regardless of success on other tasks.
     delete_model = SageMakerDeleteModelOperator(

airflow/providers/amazon/aws/example_dags/example_sqs.py

@@ -52,20 +52,20 @@ def delete_queue_fn(queue_url):
 
     create_queue = create_queue_fn()
 
-    # [START howto_sqs_operator]
+    # [START howto_operator_sqs]
     publish_to_queue = SqsPublishOperator(
         task_id='publish_to_queue',
         sqs_queue=create_queue,
         message_content="{{ task_instance }}-{{ execution_date }}",
     )
-    # [END howto_sqs_operator]
+    # [END howto_operator_sqs]
 
-    # [START howto_sqs_sensor]
+    # [START howto_sensor_sqs]
     read_from_queue = SqsSensor(
         task_id='read_from_queue',
         sqs_queue=create_queue,
     )
-    # [END howto_sqs_sensor]
+    # [END howto_sensor_sqs]
 
     delete_queue = delete_queue_fn(create_queue)
 

airflow/providers/amazon/aws/example_dags/example_step_functions.py

@@ -41,11 +41,11 @@
     )
     # [END howto_operator_step_function_start_execution]
 
-    # [START howto_operator_step_function_execution_sensor]
+    # [START howto_sensor_step_function_execution]
     wait_for_execution = StepFunctionExecutionSensor(
         task_id='wait_for_execution', execution_arn=start_execution.output
     )
-    # [END howto_operator_step_function_execution_sensor]
+    # [END howto_sensor_step_function_execution]
 
     # [START howto_operator_step_function_get_execution_output]
     get_execution_output = StepFunctionGetExecutionOutputOperator(

airflow/providers/amazon/aws/sensors/athena.py

@@ -38,7 +38,7 @@ class AthenaSensor(BaseSensorOperator):
 
     .. seealso::
         For more information on how to use this operator, take a look at the guide:
-        :ref:`howto/operator:AthenaSensor`
+        :ref:`howto/sensor:AthenaSensor`
 
 
     :param query_execution_id: query_execution_id to check the state of

airflow/providers/amazon/aws/sensors/cloud_formation.py

@@ -37,7 +37,7 @@ class CloudFormationCreateStackSensor(BaseSensorOperator):
 
     .. seealso::
         For more information on how to use this operator, take a look at the guide:
-        :ref:`howto/operator:CloudFormationCreateStackSensor`
+        :ref:`howto/sensor:CloudFormationCreateStackSensor`
 
 
     :param stack_name: The name of the stack to wait for (templated)
@@ -75,8 +75,7 @@ class CloudFormationDeleteStackSensor(BaseSensorOperator):
 
     .. seealso::
         For more information on how to use this operator, take a look at the guide:
-        :ref:`howto/operator:CloudFormationDeleteStackSensor`
-
+        :ref:`howto/sensor:CloudFormationDeleteStackSensor`
 
     :param stack_name: The name of the stack to wait for (templated)
     :param aws_conn_id: ID of the Airflow connection where credentials and extra configuration are

airflow/providers/amazon/aws/sensors/glacier.py

@@ -39,7 +39,7 @@ class GlacierJobOperationSensor(BaseSensorOperator):
 
     .. seealso::
         For more information on how to use this operator, take a look at the guide:
-        :ref:`howto/operator:GlacierJobOperationSensor`
+        :ref:`howto/sensor:GlacierJobOperationSensor`
 
     :param aws_conn_id: The reference to the AWS connection details
     :param vault_name: name of Glacier vault on which job is executed

airflow/providers/amazon/aws/sensors/sqs.py

@@ -39,7 +39,7 @@ class SqsSensor(BaseSensorOperator):
 
     .. seealso::
         For more information on how to use this sensor, take a look at the guide:
-        :ref:`howto/operator:SqsSensor`
+        :ref:`howto/sensor:SqsSensor`
 
     :param aws_conn_id: AWS connection id
     :param sqs_queue: The SQS queue url (templated)

airflow/providers/amazon/aws/sensors/step_function.py

@@ -37,7 +37,7 @@ class StepFunctionExecutionSensor(BaseSensorOperator):
 
     .. seealso::
         For more information on how to use this sensor, take a look at the guide:
-        :ref:`howto/operator:StepFunctionExecutionSensor`
+        :ref:`howto/sensor:StepFunctionExecutionSensor`
 
     :param execution_arn: execution_arn to check the state of
     :param aws_conn_id: aws connection to use, defaults to 'aws_default'

airflow/providers/amazon/aws/transfers/google_api_to_s3.py

@@ -43,6 +43,10 @@ class GoogleApiToS3Operator(BaseOperator):
     Therefore it is recommended that you use the custom Google Cloud Service Operators for working
     with the Google Cloud Platform.
 
+    .. seealso::
+        For more information on how to use this operator, take a look at the guide:
+        :ref:`howto/operator:GoogleApiToS3Operator`
+
     :param google_api_service_name: The specific API service that is being requested.
     :param google_api_service_version: The version of the API that is being requested.
     :param google_api_endpoint_path: The client libraries path to the api call's executing method.