### Discussed in https://github.com/apache/airflow/discussions/28102

<div type='discussions-op-text'>

<sup>Originally posted by **p-madduri** December  1, 2022</sup>
### Apache Airflow version

Other Airflow 2 version (please specify below)

### What happened

https://github.com/apache/airflow/blob/430e930902792fc37cdd2c517783f7dd544fbebf/airflow/providers/amazon/aws/transfers/dynamodb_to_s3.py#L39

if we use the below class at line 39:
def _convert_item_to_json_bytes(item: dict[str, Any]) -> bytes:
return (json.dumps(item) + "\n").encode("utf-8")

its throwing below error
TypeError: Object of type Decimal is not JSON serializable.

can we use

class DecimalEncoder(json.JSONEncoder):
def encode(self, obj):
if isinstance(obj, Mapping):
return '{' + ', '.join(f'{self.encode(k)}: {self.encode(v)}' for (k, v) in obj.items()) + '}'
elif isinstance(obj, Iterable) and (not isinstance(obj, str)):
return '[' + ', '.join(map(self.encode, obj)) + ']'
elif isinstance(obj, Decimal):
return f'{obj.normalize():f}' # using normalize() gets rid of trailing 0s, using ':f' prevents scientific notation
else:
print(obj)
return super().encode(obj)

and need to update the code at line 
https://github.com/apache/airflow/blob/430e930902792fc37cdd2c517783f7dd544fbebf/airflow/providers/amazon/aws/transfers/dynamodb_to_s3.py#L99

This solution is suggested in this article:
https://randomwits.com/blog/export-dynamodb-s3

Airflow version of MWAA : 2.0.2

### What you think should happen instead

mentioned in what happened section

### How to reproduce

mentioned in what happened section

### Operating System

MAC

### Versions of Apache Airflow Providers

from airflow.providers.amazon.aws.transfers.dynamodb_to_s3 import DynamoDBToS3Operator

### Deployment

MWAA

### Deployment details

n/a

### Anything else

n/a

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
</div>
responses = [
            {
                "Items": [{"a": 1}, {"b": 2}],
                "LastEvaluatedKey": "123",
            },
            {
                "Items": [{"c": 3}],
            },
        ]
        table = MagicMock()
        table.return_value.scan.side_effect = responses
        mock_aws_dynamodb_hook.return_value.get_conn.return_value.Table = table
        s3_client = MagicMock()
        s3_client.return_value.upload_file = self.mock_upload_file
        mock_s3_hook.return_value.get_conn = s3_client
        aws_conn_id = "test-conn-id"
        dynamodb_to_s3_operator = DynamoDBToS3Operator(
            task_id="dynamodb_to_s3",
            dynamodb_table_name="airflow_rocks",
            s3_bucket_name="airflow-bucket",
            file_size=4000,
            aws_conn_id=aws_conn_id,
        )
        dynamodb_to_s3_operator.execute(context={})
        assert [{"a": 1}, {"b": 2}, {"c": 3}] == self.output_queue
        mock_s3_hook.assert_called_with(aws_conn_id=aws_conn_id)
        mock_aws_dynamodb_hook.assert_called_with(aws_conn_id=aws_conn_id)