JsonGet function is not interpolated into workflow expression

[madayev]

Description
When the JsonGet function is used to get the values of PropertyFile in order to assign Estimators hyperparameter values, Sagemaker Pipeline is not able to convert this request into the expression for workflow service calls. As a result, JsonGet cannot be used to get/pass values from PropertyFile to next step components.

To reproduce

params = PropertyFile(
    name="params",
    output_name="params",
    path="params.json"
)

step_process = ProcessingStep(
    'processing_step',
    processor=processor,
    job_arguments=[...],
    outputs=[
        ...
    ],
    property_files=[params]
)

estimator = Estimator(
    image_uri=image_uri,
    instance_type=training_instance_type,
    instance_count=1,
    output_path=output_path,
    role=role,
)

estimator.set_hyperparameters(
    alpha=JsonGet(
        step=step_tune,
        property_file=params,
        json_path="alpha"
    ),
    gamma=JsonGet(
        step=step_tune,
        property_file=params,
        json_path="gamma"
    ),
    eta=JsonGet(
        step=step_tune,
        property_file=params,
        json_path="eta"
    ),
    ...
)

step_train = TrainingStep(
    name="TrainStep",
    estimator=estimator,
    inputs={
        ...
    },
)

pipeline_name = "Pipeline"
pipeline = Pipeline(
    name=pipeline_name,
    parameters=[
        ...
    ],
    steps=[step_process, step_train],
)

json.loads(pipeline.definition())

Screenshots or logs

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-9-3d1e039d5b24> in <module>
----> json.loads(pipeline.definition())

/opt/conda/lib/python3.7/site-packages/sagemaker/workflow/pipeline.py in definition(self)
    225         """Converts a request structure to string representation for workflow service calls."""
    226         request_dict = self.to_request()
--> 227         request_dict["Steps"] = interpolate(request_dict["Steps"])
    228 
    229         return json.dumps(request_dict)

/opt/conda/lib/python3.7/site-packages/sagemaker/workflow/pipeline.py in interpolate(request_obj)
    255         RequestType: The request dict with Parameter values replaced by their expression.
    256     """
--> 257     request_obj_copy = deepcopy(request_obj)
    258     return _interpolate(request_obj_copy)
    259 

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    148     copier = _deepcopy_dispatch.get(cls)
    149     if copier:
--> 150         y = copier(x, memo)
    151     else:
    152         try:

/opt/conda/lib/python3.7/copy.py in _deepcopy_list(x, memo, deepcopy)
    214     append = y.append
    215     for a in x:
--> 216         append(deepcopy(a, memo))
    217     return y
    218 d[list] = _deepcopy_list

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    148     copier = _deepcopy_dispatch.get(cls)
    149     if copier:
--> 150         y = copier(x, memo)
    151     else:
    152         try:

/opt/conda/lib/python3.7/copy.py in _deepcopy_dict(x, memo, deepcopy)
    239     memo[id(x)] = y
    240     for key, value in x.items():
--> 241         y[deepcopy(key, memo)] = deepcopy(value, memo)
    242     return y
    243 d[dict] = _deepcopy_dict

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    148     copier = _deepcopy_dispatch.get(cls)
    149     if copier:
--> 150         y = copier(x, memo)
    151     else:
    152         try:

/opt/conda/lib/python3.7/copy.py in _deepcopy_dict(x, memo, deepcopy)
    239     memo[id(x)] = y
    240     for key, value in x.items():
--> 241         y[deepcopy(key, memo)] = deepcopy(value, memo)
    242     return y
    243 d[dict] = _deepcopy_dict

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    148     copier = _deepcopy_dispatch.get(cls)
    149     if copier:
--> 150         y = copier(x, memo)
    151     else:
    152         try:

/opt/conda/lib/python3.7/copy.py in _deepcopy_dict(x, memo, deepcopy)
    239     memo[id(x)] = y
    240     for key, value in x.items():
--> 241         y[deepcopy(key, memo)] = deepcopy(value, memo)
    242     return y
    243 d[dict] = _deepcopy_dict

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    178                     y = x
    179                 else:
--> 180                     y = _reconstruct(x, memo, *rv)
    181 
    182     # If is its own copy, don't memoize.

/opt/conda/lib/python3.7/copy.py in _reconstruct(x, memo, func, args, state, listiter, dictiter, deepcopy)
    279     if state is not None:
    280         if deep:
--> 281             state = deepcopy(state, memo)
    282         if hasattr(y, '__setstate__'):
    283             y.__setstate__(state)

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    148     copier = _deepcopy_dispatch.get(cls)
    149     if copier:
--> 150         y = copier(x, memo)
    151     else:
    152         try:

/opt/conda/lib/python3.7/copy.py in _deepcopy_dict(x, memo, deepcopy)
    239     memo[id(x)] = y
    240     for key, value in x.items():
--> 241         y[deepcopy(key, memo)] = deepcopy(value, memo)
    242     return y
    243 d[dict] = _deepcopy_dict

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    178                     y = x
    179                 else:
--> 180                     y = _reconstruct(x, memo, *rv)
    181 
    182     # If is its own copy, don't memoize.

/opt/conda/lib/python3.7/copy.py in _reconstruct(x, memo, func, args, state, listiter, dictiter, deepcopy)
    279     if state is not None:
    280         if deep:
--> 281             state = deepcopy(state, memo)
    282         if hasattr(y, '__setstate__'):
    283             y.__setstate__(state)

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    148     copier = _deepcopy_dispatch.get(cls)
    149     if copier:
--> 150         y = copier(x, memo)
    151     else:
    152         try:

/opt/conda/lib/python3.7/copy.py in _deepcopy_dict(x, memo, deepcopy)
    239     memo[id(x)] = y
    240     for key, value in x.items():
--> 241         y[deepcopy(key, memo)] = deepcopy(value, memo)
    242     return y
    243 d[dict] = _deepcopy_dict

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    178                     y = x
    179                 else:
--> 180                     y = _reconstruct(x, memo, *rv)
    181 
    182     # If is its own copy, don't memoize.

/opt/conda/lib/python3.7/copy.py in _reconstruct(x, memo, func, args, state, listiter, dictiter, deepcopy)
    279     if state is not None:
    280         if deep:
--> 281             state = deepcopy(state, memo)
    282         if hasattr(y, '__setstate__'):
    283             y.__setstate__(state)

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    148     copier = _deepcopy_dispatch.get(cls)
    149     if copier:
--> 150         y = copier(x, memo)
    151     else:
    152         try:

/opt/conda/lib/python3.7/copy.py in _deepcopy_dict(x, memo, deepcopy)
    239     memo[id(x)] = y
    240     for key, value in x.items():
--> 241         y[deepcopy(key, memo)] = deepcopy(value, memo)
    242     return y
    243 d[dict] = _deepcopy_dict

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    178                     y = x
    179                 else:
--> 180                     y = _reconstruct(x, memo, *rv)
    181 
    182     # If is its own copy, don't memoize.

/opt/conda/lib/python3.7/copy.py in _reconstruct(x, memo, func, args, state, listiter, dictiter, deepcopy)
    279     if state is not None:
    280         if deep:
--> 281             state = deepcopy(state, memo)
    282         if hasattr(y, '__setstate__'):
    283             y.__setstate__(state)

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    148     copier = _deepcopy_dispatch.get(cls)
    149     if copier:
--> 150         y = copier(x, memo)
    151     else:
    152         try:

/opt/conda/lib/python3.7/copy.py in _deepcopy_dict(x, memo, deepcopy)
    239     memo[id(x)] = y
    240     for key, value in x.items():
--> 241         y[deepcopy(key, memo)] = deepcopy(value, memo)
    242     return y
    243 d[dict] = _deepcopy_dict

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    178                     y = x
    179                 else:
--> 180                     y = _reconstruct(x, memo, *rv)
    181 
    182     # If is its own copy, don't memoize.

/opt/conda/lib/python3.7/copy.py in _reconstruct(x, memo, func, args, state, listiter, dictiter, deepcopy)
    279     if state is not None:
    280         if deep:
--> 281             state = deepcopy(state, memo)
    282         if hasattr(y, '__setstate__'):
    283             y.__setstate__(state)

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    148     copier = _deepcopy_dispatch.get(cls)
    149     if copier:
--> 150         y = copier(x, memo)
    151     else:
    152         try:

/opt/conda/lib/python3.7/copy.py in _deepcopy_dict(x, memo, deepcopy)
    239     memo[id(x)] = y
    240     for key, value in x.items():
--> 241         y[deepcopy(key, memo)] = deepcopy(value, memo)
    242     return y
    243 d[dict] = _deepcopy_dict

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    178                     y = x
    179                 else:
--> 180                     y = _reconstruct(x, memo, *rv)
    181 
    182     # If is its own copy, don't memoize.

/opt/conda/lib/python3.7/copy.py in _reconstruct(x, memo, func, args, state, listiter, dictiter, deepcopy)
    279     if state is not None:
    280         if deep:
--> 281             state = deepcopy(state, memo)
    282         if hasattr(y, '__setstate__'):
    283             y.__setstate__(state)

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    148     copier = _deepcopy_dispatch.get(cls)
    149     if copier:
--> 150         y = copier(x, memo)
    151     else:
    152         try:

/opt/conda/lib/python3.7/copy.py in _deepcopy_dict(x, memo, deepcopy)
    239     memo[id(x)] = y
    240     for key, value in x.items():
--> 241         y[deepcopy(key, memo)] = deepcopy(value, memo)
    242     return y
    243 d[dict] = _deepcopy_dict

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    178                     y = x
    179                 else:
--> 180                     y = _reconstruct(x, memo, *rv)
    181 
    182     # If is its own copy, don't memoize.

/opt/conda/lib/python3.7/copy.py in _reconstruct(x, memo, func, args, state, listiter, dictiter, deepcopy)
    279     if state is not None:
    280         if deep:
--> 281             state = deepcopy(state, memo)
    282         if hasattr(y, '__setstate__'):
    283             y.__setstate__(state)

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    148     copier = _deepcopy_dispatch.get(cls)
    149     if copier:
--> 150         y = copier(x, memo)
    151     else:
    152         try:

/opt/conda/lib/python3.7/copy.py in _deepcopy_dict(x, memo, deepcopy)
    239     memo[id(x)] = y
    240     for key, value in x.items():
--> 241         y[deepcopy(key, memo)] = deepcopy(value, memo)
    242     return y
    243 d[dict] = _deepcopy_dict

/opt/conda/lib/python3.7/copy.py in deepcopy(x, memo, _nil)
    167                     reductor = getattr(x, "__reduce_ex__", None)
    168                     if reductor:
--> 169                         rv = reductor(4)
    170                     else:
    171                         reductor = getattr(x, "__reduce__", None)

TypeError: can't pickle _thread.lock objects

Expected behavior
Successfully convert the request to workflow service calls and get values from PropertyFile at runtime.

System information

• SageMaker Python SDK version:
• Framework name (eg. PyTorch) or algorithm (eg. KMeans):
• Framework version:
• Python version:
• CPU or GPU:
• Custom Docker image (Y/N):

Additional context
It's possible to bypass the error by overriding the deepcopy function, so it returns a new instance of JsonGet.

[kirit93]

@madayev - can you share how you were able to do this - " It's possible to bypass the error by overriding the deepcopy function, so it returns a new instance of JsonGet."
Were you able to successfully run a pipeline with that workaround?