DynamoDBToS3Operator - Add feature to export table to a point in time

[utkarsharma2]

DynamoDBToS3Operator - Add a feature to export the table to a point in time.

closes: #28830

[utkarsharma2]

@eladkal @o-nikolas Please review

[utkarsharma2]

FYI - Test cases are failing because of the following issue - #30613

[potiuk]

Needs conflict resolving too.

[ferruzzi]

Left a couple of comments and non-blocking suggestion, but I like how this turned out. The PascalCase suggestion needs to be fixed, the other two are just thoughts you can feel free to revise or ignore.

I really appreciate your patience with the waiter shenanigans, we both learned something on that!

[ferruzzi]

Nice 👍

[ferruzzi]

❤️

[ferruzzi]

If the CI is happy, I'm happy 👍 Thanks for sticking with this one.

[potiuk]

CI is not happy. @utkarsharma2 -> you need to fix those .

[utkarsharma2]

Hey @potiuk, CI is happy now :)

[ferruzzi]

@utkarsharma2 Did you actually try this live? After the merge, the system test fails because this addition is missing the file_size paramter, and after adding that in it fails with the message

ERROR    airflow.task:taskinstance.py:1900 Task failed with exception
Traceback (most recent call last):
  File "/opt/airflow/airflow/providers/amazon/aws/transfers/dynamodb_to_s3.py", line 142, in execute
    self._export_table_to_point_in_time()
  File "/opt/airflow/airflow/providers/amazon/aws/transfers/dynamodb_to_s3.py", line 157, in _export_table_to_point_in_time
    ExportFormat=self.export_format,
  File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 530, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 960, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (ValidationException) when calling the ExportTableToPointInTime operation: One or more parameter values were invalid: tableArn is not a valid ARN

Here you are passingdynamodb_table_name instead of the ARN. I'm not sure how this would have worked, and I probably should have noticed that in the review.

[utkarsharma2]

@ferruzzi Yes I did test it locally, I was passing table ARN in place of the table name, and file_size but forgot to add it in the example :/
Below is the code that worked for me.

  dynamodb_to_s3_operator = DynamoDBToS3Operator(
      task_id="dynamodb_to_s3",
      dynamodb_table_name="arn:aws:dynamodb:us-east-1:633294268925:table/test",
      s3_bucket_name="tmp9",
      file_size=4000,
      export_time=datetime.now(),
      aws_conn_id="aws_default",
      s3_key_prefix="test"
  )

[utkarsharma2]

@ferruzzi @o-nikolas Raised a new PR with the Fix Please review - #31142

[ferruzzi]

Cool, thanks! I'll take a look later today when I get a moment 👍