### 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>
while True:
            response = table.scan(**scan_kwargs)
            items = response["Items"]
            for item in items:
                temp_file.write(self.process_func(item))
            if "LastEvaluatedKey" not in response:
                
                break
            last_evaluated_key = response["LastEvaluatedKey"]
            scan_kwargs["ExclusiveStartKey"] = last_evaluated_key
            
            if getsize(temp_file.name) >= self.file_size:
                _upload_file_to_s3(temp_file, self.s3_bucket_name, self.s3_key_prefix, self.aws_conn_id)
                temp_file.close()
                temp_file = NamedTemporaryFile()
        return temp_file