Type Error while using dynamodb_to_s3 operator

[potiuk]

Discussed in #28102

^{Originally posted by p-madduri December 1, 2022}

Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

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

Line 39 in 430e930

def _convert_item_to_json_bytes(item: dict[str, Any]) -> bytes:

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

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

Line 99 in 430e930

process_func: Callable[[dict[str, Any]], bytes] = _convert_item_to_json_bytes,

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?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[snjypl]

@potiuk i was wondering if this could be a solution.

import json
from decimal import Decimal


class JSONEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, Decimal):
              return float(obj)
        return json.JSONEncoder.default(self, obj)


for i in ["102938.3043847474",1.010001,10,"100","1E-128",1E-128]:
    org = Decimal(i)
    encoded = json.dumps(org,cls=JSONEncoder)
    decoded = json.loads(encoded,parse_float=Decimal)
    
    
    print(f"{org=}")
    print(f"{encoded=}")
    print(f"{decoded=}")
    print(" "*5)

output:

org=Decimal('102938.3043847474')
encoded='102938.3043847474'
decoded=Decimal('102938.3043847474')

org=Decimal('1.01000099999999992661514625069685280323028564453125')
encoded='1.010001'
decoded=Decimal('1.010001')

org=Decimal('10')
encoded='10.0'
decoded=Decimal('10.0')

org=Decimal('100')
encoded='100.0'
decoded=Decimal('100.0')

org=Decimal('1E-128')
encoded='1e-128'
decoded=Decimal('1E-128')

org=Decimal('1.00000000000000005401408859568103309052834145426594802430862986593345890744772751462514653740544965836726232958009946209108101694078484369205457039971496303344480117025912844198468177541980552987602212529712035107713678381275316882208085186144350914663534730232928215257050045129264726062873538499086450376880826074188490792948869056999683380126953125E-128')
encoded='1e-128'
decoded=Decimal('1E-128')

there is something similar in airflow uitls

airflow/airflow/utils/json.py

Lines 77 to 83 in 64deada

	elif isinstance(obj, Decimal):
	_, _, exponent = obj.as_tuple()
	if exponent >= 0: # No digits after the decimal point.
	return int(obj)
	# Technically lossy due to floating point errors, but the best we
	# can do without implementing a custom encode function.
	return float(obj)

[potiuk]

Make a PR - It's far easier to reason about it when you just make a PR and run tests.