You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "Polber (via GitHub)" <gi...@apache.org> on 2023/10/27 16:33:06 UTC
[PR] Jkinard/fix jsobject [beam]
Polber opened a new pull request, #29169:
URL: https://github.com/apache/beam/pull/29169
Add JSObject python conversion to YAML Javascript UDF Mapping transforms.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1375070556
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
+ for idx in range(len(row_list)):
+ row_list[idx] = row_to_dict(row_list[idx])
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ PyJsArray,
+ PyJsArrayBuffer,
+ PyJsInt8Array,
+ PyJsUint8Array,
+ PyJsUint8ClampedArray,
+ PyJsInt16Array,
+ PyJsUint16Array,
+ PyJsInt32Array,
+ PyJsUint32Array,
+ PyJsFloat32Array,
+ PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (PyJsNumber, PyJsString, PyJsBoolean)):
+ obj = to_python(obj)
+ elif isinstance(obj, js_array_type):
+ return [_js_object_to_py_object(value) for value in obj.to_list()]
+ elif isinstance(obj, PyJsDate):
+ obj = obj.to_utc_dt()
+ elif isinstance(obj, (PyJsNull, PyJsUndefined)):
+ return None
+ elif isinstance(obj, PyJsError):
+ raise RuntimeError(obj['message'])
+ elif isinstance(obj, PyJsObject):
+ return {
+ key: _js_object_to_py_object(value['value'])
+ for key,
+ value in obj.own.items()
+ }
+ elif isinstance(obj, JsObjectWrapper):
+ return _js_object_to_py_object(obj._obj)
+
+ return obj
+
+ def _catch_js_errors(func):
+ try:
+ result = func()
+ except JsException as e:
+ result = getattr(e, 'mes')
+ return result
+
if expression:
args = ', '.join(original_fields)
js_func = f'function fn({args}) {{return ({expression})}}'
- js_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
- return lambda __row__: js_callable(*__row__._asdict().values())
+ js_expr_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
+ fn = lambda __row__: lambda: js_expr_callable(
+ *row_to_dict(__row__).values())
elif callable:
js_callable = _CustomJsObjectWrapper(js2py.eval_js(callable))
- return lambda __row__: js_callable(__row__._asdict())
+ fn = lambda __row__: lambda: js_callable(row_to_dict(__row__))
else:
if not path.endswith('.js'):
raise ValueError(f'File "{path}" is not a valid .js file.')
udf_code = FileSystems.open(path).read().decode()
js = js2py.EvalJs()
js.eval(udf_code)
- js_callable = _CustomJsObjectWrapper(getattr(js, name))
- return lambda __row__: js_callable(__row__._asdict())
+ js_file_callable = _CustomJsObjectWrapper(getattr(js, name))
+ fn = lambda __row__: lambda: js_file_callable(row_to_dict(__row__))
+
+ return lambda __row__: dicts_to_rows(
+ _js_object_to_py_object(_catch_js_errors(fn(__row__))))
Review Comment:
It does propagate. It extracts a `PyJsError` object and the error is thrown in `_js_object_to_py_object()` which is caught by the existing error handling framework. I thought it was a bit cleaner of an error message because if you have a custom error/exception in the JS script, it gets a bit hidden in the stack trace of JSException as opposed to being the top-level error
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #29169:
URL: https://github.com/apache/beam/pull/29169#issuecomment-1783233079
We're currently processing your upload. This comment will be updated when the results are available.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1375055340
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1375045991
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -27,8 +27,31 @@
from typing import Union
import js2py
+from js2py.base import JsObjectWrapper
+from js2py.base import PyJsArray
+from js2py.base import PyJsArrayBuffer
+from js2py.base import PyJsBoolean
+from js2py.base import PyJsError
+from js2py.base import PyJsFloat32Array
+from js2py.base import PyJsFloat64Array
+from js2py.base import PyJsInt16Array
+from js2py.base import PyJsInt32Array
+from js2py.base import PyJsInt8Array
+from js2py.base import PyJsNull
+from js2py.base import PyJsNumber
+from js2py.base import PyJsObject
+from js2py.base import PyJsString
+from js2py.base import PyJsUint16Array
+from js2py.base import PyJsUint32Array
+from js2py.base import PyJsUint8Array
+from js2py.base import PyJsUint8ClampedArray
+from js2py.base import PyJsUndefined
+from js2py.base import to_python
+from js2py.constructors.jsdate import PyJsDate
+from js2py.internals.simplex import JsException
import apache_beam as beam
+from apache_beam import Row
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1375060075
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
+ for idx in range(len(row_list)):
+ row_list[idx] = row_to_dict(row_list[idx])
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ PyJsArray,
+ PyJsArrayBuffer,
+ PyJsInt8Array,
+ PyJsUint8Array,
+ PyJsUint8ClampedArray,
+ PyJsInt16Array,
+ PyJsUint16Array,
+ PyJsInt32Array,
+ PyJsUint32Array,
+ PyJsFloat32Array,
+ PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (PyJsNumber, PyJsString, PyJsBoolean)):
+ obj = to_python(obj)
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "robertwb (via GitHub)" <gi...@apache.org>.
robertwb commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1375109787
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +79,85 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if ((isinstance(row, tuple) and hasattr(row, '_asdict')) or
+ isinstance(row, beam.Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ return {key: row_to_dict(value) for key, value in row.items()}
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ return [row_to_dict(value) for value in list(row)]
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ base.PyJsArray,
+ base.PyJsArrayBuffer,
+ base.PyJsInt8Array,
+ base.PyJsUint8Array,
+ base.PyJsUint8ClampedArray,
+ base.PyJsInt16Array,
+ base.PyJsUint16Array,
+ base.PyJsInt32Array,
+ base.PyJsUint32Array,
+ base.PyJsFloat32Array,
+ base.PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (base.PyJsNumber, base.PyJsString, base.PyJsBoolean)):
+ return base.to_python(obj)
+ elif isinstance(obj, js_array_type):
+ return [_js_object_to_py_object(value) for value in obj.to_list()]
+ elif isinstance(obj, jsdate.PyJsDate):
+ return obj.to_utc_dt()
+ elif isinstance(obj, (base.PyJsNull, base.PyJsUndefined)):
+ return None
+ elif isinstance(obj, base.PyJsError):
+ raise RuntimeError(obj['message'])
+ elif isinstance(obj, base.PyJsObject):
+ return {
+ key: _js_object_to_py_object(value['value'])
+ for (key, value) in obj.own.items()
+ }
+ elif isinstance(obj, base.JsObjectWrapper):
+ return _js_object_to_py_object(obj._obj)
+
+ return obj
+
+ def _catch_js_errors(func):
+ try:
+ result = func()
+ except simplex.JsException as e:
+ result = getattr(e, 'mes')
Review Comment:
why not `e.mes`?
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +79,85 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if ((isinstance(row, tuple) and hasattr(row, '_asdict')) or
+ isinstance(row, beam.Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ return {key: row_to_dict(value) for key, value in row.items()}
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ return [row_to_dict(value) for value in list(row)]
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ base.PyJsArray,
+ base.PyJsArrayBuffer,
+ base.PyJsInt8Array,
+ base.PyJsUint8Array,
+ base.PyJsUint8ClampedArray,
+ base.PyJsInt16Array,
+ base.PyJsUint16Array,
+ base.PyJsInt32Array,
+ base.PyJsUint32Array,
+ base.PyJsFloat32Array,
+ base.PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (base.PyJsNumber, base.PyJsString, base.PyJsBoolean)):
+ return base.to_python(obj)
+ elif isinstance(obj, js_array_type):
+ return [_js_object_to_py_object(value) for value in obj.to_list()]
+ elif isinstance(obj, jsdate.PyJsDate):
+ return obj.to_utc_dt()
+ elif isinstance(obj, (base.PyJsNull, base.PyJsUndefined)):
+ return None
+ elif isinstance(obj, base.PyJsError):
+ raise RuntimeError(obj['message'])
+ elif isinstance(obj, base.PyJsObject):
+ return {
+ key: _js_object_to_py_object(value['value'])
+ for (key, value) in obj.own.items()
+ }
+ elif isinstance(obj, base.JsObjectWrapper):
+ return _js_object_to_py_object(obj._obj)
+
+ return obj
+
+ def _catch_js_errors(func):
+ try:
+ result = func()
+ except simplex.JsException as e:
+ result = getattr(e, 'mes')
+ return result
+
if expression:
args = ', '.join(original_fields)
js_func = f'function fn({args}) {{return ({expression})}}'
- js_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
- return lambda __row__: js_callable(*__row__._asdict().values())
+ js_expr_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
+ fn = lambda __row__: lambda: js_expr_callable(
+ *row_to_dict(__row__).values())
Review Comment:
Does js_expr_callable take keyword arguments so you don't have to rely on the ordering of `row_to_dict(__row__).values()`? (I think we'll be OK, as long as __row__ has exactly the expected schema, but maybe it is still compatible but has extra fields?)
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
+ for idx in range(len(row_list)):
+ row_list[idx] = row_to_dict(row_list[idx])
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ PyJsArray,
+ PyJsArrayBuffer,
+ PyJsInt8Array,
+ PyJsUint8Array,
+ PyJsUint8ClampedArray,
+ PyJsInt16Array,
+ PyJsUint16Array,
+ PyJsInt32Array,
+ PyJsUint32Array,
+ PyJsFloat32Array,
+ PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (PyJsNumber, PyJsString, PyJsBoolean)):
+ obj = to_python(obj)
+ elif isinstance(obj, js_array_type):
+ return [_js_object_to_py_object(value) for value in obj.to_list()]
+ elif isinstance(obj, PyJsDate):
+ obj = obj.to_utc_dt()
+ elif isinstance(obj, (PyJsNull, PyJsUndefined)):
+ return None
+ elif isinstance(obj, PyJsError):
+ raise RuntimeError(obj['message'])
+ elif isinstance(obj, PyJsObject):
+ return {
+ key: _js_object_to_py_object(value['value'])
+ for key,
+ value in obj.own.items()
+ }
+ elif isinstance(obj, JsObjectWrapper):
+ return _js_object_to_py_object(obj._obj)
+
+ return obj
+
+ def _catch_js_errors(func):
+ try:
+ result = func()
+ except JsException as e:
+ result = getattr(e, 'mes')
+ return result
+
if expression:
args = ', '.join(original_fields)
js_func = f'function fn({args}) {{return ({expression})}}'
- js_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
- return lambda __row__: js_callable(*__row__._asdict().values())
+ js_expr_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
+ fn = lambda __row__: lambda: js_expr_callable(
+ *row_to_dict(__row__).values())
elif callable:
js_callable = _CustomJsObjectWrapper(js2py.eval_js(callable))
- return lambda __row__: js_callable(__row__._asdict())
+ fn = lambda __row__: lambda: js_callable(row_to_dict(__row__))
else:
if not path.endswith('.js'):
raise ValueError(f'File "{path}" is not a valid .js file.')
udf_code = FileSystems.open(path).read().decode()
js = js2py.EvalJs()
js.eval(udf_code)
- js_callable = _CustomJsObjectWrapper(getattr(js, name))
- return lambda __row__: js_callable(__row__._asdict())
+ js_file_callable = _CustomJsObjectWrapper(getattr(js, name))
+ fn = lambda __row__: lambda: js_file_callable(row_to_dict(__row__))
+
+ return lambda __row__: dicts_to_rows(
+ _js_object_to_py_object(_catch_js_errors(fn(__row__))))
Review Comment:
Catching an exception, turning it into a value, and then re-raising that value elsewhere makes things a lot harder to follow though. E.g. you catch simplex.JsException, get the `mes` attribute (which is presumably of type `PyJsError` so you can detect it later?, and then raise during conversion?
If you want a simpler stack trace, you can do something like
```
try:
return func()
except simplex.JsException as exn:
raise RuntimeError(f"Error evaluating javascript expression: {exn.msg}") from exn
```
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +79,85 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if ((isinstance(row, tuple) and hasattr(row, '_asdict')) or
+ isinstance(row, beam.Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ return {key: row_to_dict(value) for key, value in row.items()}
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ return [row_to_dict(value) for value in list(row)]
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ base.PyJsArray,
+ base.PyJsArrayBuffer,
+ base.PyJsInt8Array,
+ base.PyJsUint8Array,
+ base.PyJsUint8ClampedArray,
+ base.PyJsInt16Array,
+ base.PyJsUint16Array,
+ base.PyJsInt32Array,
+ base.PyJsUint32Array,
+ base.PyJsFloat32Array,
+ base.PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (base.PyJsNumber, base.PyJsString, base.PyJsBoolean)):
+ return base.to_python(obj)
+ elif isinstance(obj, js_array_type):
+ return [_js_object_to_py_object(value) for value in obj.to_list()]
+ elif isinstance(obj, jsdate.PyJsDate):
+ return obj.to_utc_dt()
+ elif isinstance(obj, (base.PyJsNull, base.PyJsUndefined)):
+ return None
+ elif isinstance(obj, base.PyJsError):
+ raise RuntimeError(obj['message'])
+ elif isinstance(obj, base.PyJsObject):
+ return {
+ key: _js_object_to_py_object(value['value'])
+ for (key, value) in obj.own.items()
+ }
+ elif isinstance(obj, base.JsObjectWrapper):
+ return _js_object_to_py_object(obj._obj)
+
+ return obj
+
+ def _catch_js_errors(func):
+ try:
+ result = func()
+ except simplex.JsException as e:
+ result = getattr(e, 'mes')
+ return result
+
if expression:
args = ', '.join(original_fields)
js_func = f'function fn({args}) {{return ({expression})}}'
- js_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
- return lambda __row__: js_callable(*__row__._asdict().values())
+ js_expr_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
+ fn = lambda __row__: lambda: js_expr_callable(
Review Comment:
No need for the double-underscore __row__ as you're not worried about polluting a namespace here.
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +79,85 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if ((isinstance(row, tuple) and hasattr(row, '_asdict')) or
+ isinstance(row, beam.Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ return {key: row_to_dict(value) for key, value in row.items()}
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ return [row_to_dict(value) for value in list(row)]
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ base.PyJsArray,
+ base.PyJsArrayBuffer,
+ base.PyJsInt8Array,
+ base.PyJsUint8Array,
+ base.PyJsUint8ClampedArray,
+ base.PyJsInt16Array,
+ base.PyJsUint16Array,
+ base.PyJsInt32Array,
+ base.PyJsUint32Array,
+ base.PyJsFloat32Array,
+ base.PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (base.PyJsNumber, base.PyJsString, base.PyJsBoolean)):
+ return base.to_python(obj)
+ elif isinstance(obj, js_array_type):
+ return [_js_object_to_py_object(value) for value in obj.to_list()]
+ elif isinstance(obj, jsdate.PyJsDate):
+ return obj.to_utc_dt()
+ elif isinstance(obj, (base.PyJsNull, base.PyJsUndefined)):
+ return None
+ elif isinstance(obj, base.PyJsError):
+ raise RuntimeError(obj['message'])
+ elif isinstance(obj, base.PyJsObject):
+ return {
+ key: _js_object_to_py_object(value['value'])
+ for (key, value) in obj.own.items()
+ }
+ elif isinstance(obj, base.JsObjectWrapper):
+ return _js_object_to_py_object(obj._obj)
+
+ return obj
+
+ def _catch_js_errors(func):
+ try:
+ result = func()
+ except simplex.JsException as e:
+ result = getattr(e, 'mes')
+ return result
+
if expression:
args = ', '.join(original_fields)
js_func = f'function fn({args}) {{return ({expression})}}'
- js_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
- return lambda __row__: js_callable(*__row__._asdict().values())
+ js_expr_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
+ fn = lambda __row__: lambda: js_expr_callable(
+ *row_to_dict(__row__).values())
elif callable:
js_callable = _CustomJsObjectWrapper(js2py.eval_js(callable))
- return lambda __row__: js_callable(__row__._asdict())
+ fn = lambda __row__: lambda: js_callable(row_to_dict(__row__))
else:
if not path.endswith('.js'):
raise ValueError(f'File "{path}" is not a valid .js file.')
udf_code = FileSystems.open(path).read().decode()
js = js2py.EvalJs()
js.eval(udf_code)
- js_callable = _CustomJsObjectWrapper(getattr(js, name))
- return lambda __row__: js_callable(__row__._asdict())
+ js_file_callable = _CustomJsObjectWrapper(getattr(js, name))
+ fn = lambda __row__: lambda: js_file_callable(row_to_dict(__row__))
Review Comment:
It makes it a bit hard to follow to have fn be a lambda that takes a row and returns a callable. If you want this pattern, this could likely be better done with function decorators. But there's also redundancy here as well. I think it'd be cleaner to do something like
```
fn = [your previous logic defining a js -> js evaluation]
def wrapper(row):
try:
return dicts_to_rows(_js_object_to_py_object(fn(row_to_dict(row))))
except simplex.JsException as exn:
raise RuntimeError(f"Error evaluating javascript expression: {exn.msg}") from exn
return wrapper
```
If you want to split up the errors you could also do
```
def wrapper(row):
row_as_dict = row_to_dict(row)
try:
js_result = fn(row_as_dict)
except simplex.JsException as exn:
raise RuntimeError(f"Error evaluating javascript expression: {exn.msg}") from exn
return dicts_to_rows(_js_object_to_py_object(js_result))
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Jkinard/fix jsobject [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on PR #29169:
URL: https://github.com/apache/beam/pull/29169#issuecomment-1783198371
R: @robertwb
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "robertwb (via GitHub)" <gi...@apache.org>.
robertwb merged PR #29169:
URL: https://github.com/apache/beam/pull/29169
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1375060544
##########
sdks/python/apache_beam/yaml/yaml_udf_test.py:
##########
@@ -60,16 +66,37 @@ def test_map_to_fields_filter_inline_js(self):
language: javascript
fields:
label:
- callable: "function label_map(x) {return x.label + 'x'}"
+ callable: |
+ function label_map(x) {
Review Comment:
This is testing callable specifically. I wanted to make tests that only use 1 of the 3 types (expression, callable, file) so that the failures are granular.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "robertwb (via GitHub)" <gi...@apache.org>.
robertwb commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1375106335
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -100,16 +80,13 @@ def __setstate__(self, state):
def row_to_dict(row):
- if (str(type(row).__name__).startswith('BeamSchema_') or
- isinstance(row, Row)):
+ if ((isinstance(row, tuple) and hasattr(row, '_asdict')) or
+ isinstance(row, beam.Row)):
row = row._asdict()
if isinstance(row, dict):
- for key, value in row.items():
- row[key] = row_to_dict(value)
+ return {key: row_to_dict(value) for key, value in row.items()}
elif not isinstance(row, str) and isinstance(row, Iterable):
- row_list = list(row)
- for idx in range(len(row_list)):
- row_list[idx] = row_to_dict(row_list[idx])
+ return [row_to_dict(value) for value in list(row)]
return row
Review Comment:
Nit. I'd put the last return in an else clause (as it's a "mutually exclusive cases" kind of setup.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "robertwb (via GitHub)" <gi...@apache.org>.
robertwb commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1374875434
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -27,8 +27,31 @@
from typing import Union
import js2py
+from js2py.base import JsObjectWrapper
+from js2py.base import PyJsArray
+from js2py.base import PyJsArrayBuffer
+from js2py.base import PyJsBoolean
+from js2py.base import PyJsError
+from js2py.base import PyJsFloat32Array
+from js2py.base import PyJsFloat64Array
+from js2py.base import PyJsInt16Array
+from js2py.base import PyJsInt32Array
+from js2py.base import PyJsInt8Array
+from js2py.base import PyJsNull
+from js2py.base import PyJsNumber
+from js2py.base import PyJsObject
+from js2py.base import PyJsString
+from js2py.base import PyJsUint16Array
+from js2py.base import PyJsUint32Array
+from js2py.base import PyJsUint8Array
+from js2py.base import PyJsUint8ClampedArray
+from js2py.base import PyJsUndefined
+from js2py.base import to_python
+from js2py.constructors.jsdate import PyJsDate
+from js2py.internals.simplex import JsException
import apache_beam as beam
+from apache_beam import Row
Review Comment:
Same. beam.Row is sufficient.
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -27,8 +27,31 @@
from typing import Union
import js2py
+from js2py.base import JsObjectWrapper
Review Comment:
We generally try to follow https://google.github.io/styleguide/pyguide.html#22-imports (import the module, not each individual thing). You might want to import base as a more descriptive name though.
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
+ for idx in range(len(row_list)):
+ row_list[idx] = row_to_dict(row_list[idx])
+ return row
Review Comment:
This is a case where it's probably clearer to put the return in each clause than re-assign row and return the mutated value a the end. An else clause can return the input itself. (Technically, it might not be a row, so value might be a better name.)
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
+ for idx in range(len(row_list)):
+ row_list[idx] = row_to_dict(row_list[idx])
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ PyJsArray,
+ PyJsArrayBuffer,
+ PyJsInt8Array,
+ PyJsUint8Array,
+ PyJsUint8ClampedArray,
+ PyJsInt16Array,
+ PyJsUint16Array,
+ PyJsInt32Array,
+ PyJsUint32Array,
+ PyJsFloat32Array,
+ PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (PyJsNumber, PyJsString, PyJsBoolean)):
+ obj = to_python(obj)
+ elif isinstance(obj, js_array_type):
+ return [_js_object_to_py_object(value) for value in obj.to_list()]
+ elif isinstance(obj, PyJsDate):
+ obj = obj.to_utc_dt()
+ elif isinstance(obj, (PyJsNull, PyJsUndefined)):
+ return None
+ elif isinstance(obj, PyJsError):
+ raise RuntimeError(obj['message'])
+ elif isinstance(obj, PyJsObject):
Review Comment:
On this note, it's probably ideal if we can get an expected schema ahead of time, and then use that to covert. (See, e.g. how PubSub or Avro write works).
##########
sdks/python/apache_beam/yaml/yaml_udf_test.py:
##########
@@ -60,16 +66,37 @@ def test_map_to_fields_filter_inline_js(self):
language: javascript
fields:
label:
- callable: "function label_map(x) {return x.label + 'x'}"
+ callable: |
+ function label_map(x) {
Review Comment:
These'd be more natural as expressions, right? Or maybe that's not what you're trying to test here.
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
+ for idx in range(len(row_list)):
+ row_list[idx] = row_to_dict(row_list[idx])
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ PyJsArray,
+ PyJsArrayBuffer,
+ PyJsInt8Array,
+ PyJsUint8Array,
+ PyJsUint8ClampedArray,
+ PyJsInt16Array,
+ PyJsUint16Array,
+ PyJsInt32Array,
+ PyJsUint32Array,
+ PyJsFloat32Array,
+ PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (PyJsNumber, PyJsString, PyJsBoolean)):
+ obj = to_python(obj)
+ elif isinstance(obj, js_array_type):
+ return [_js_object_to_py_object(value) for value in obj.to_list()]
+ elif isinstance(obj, PyJsDate):
+ obj = obj.to_utc_dt()
+ elif isinstance(obj, (PyJsNull, PyJsUndefined)):
+ return None
+ elif isinstance(obj, PyJsError):
+ raise RuntimeError(obj['message'])
+ elif isinstance(obj, PyJsObject):
Review Comment:
So here's where some discretion is needed. Should this be a Row or a Mapping? E.g. a literal return of {a: 1, b: "foo"} is probably intended to be a Row.
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
Review Comment:
Not all schemas start with BeamSchema (e.g. user defined named tuples). The test here is whether it's a named tuple. You could do isinstance(tuple) and hasattr(_asdict).
##########
sdks/python/apache_beam/yaml/yaml_udf_test.py:
##########
@@ -25,20 +25,26 @@
from apache_beam.options import pipeline_options
from apache_beam.testing.util import assert_that
from apache_beam.testing.util import equal_to
+from apache_beam.yaml.yaml_mapping import row_to_dict
+from apache_beam.yaml.yaml_provider import dicts_to_rows
from apache_beam.yaml.yaml_transform import YamlTransform
def AsRows():
- return beam.Map(lambda named_tuple: beam.Row(**named_tuple._asdict()))
+ return beam.Map(lambda named_tuple: dicts_to_rows(row_to_dict(named_tuple)))
class YamlUDFMappingTest(unittest.TestCase):
def __init__(self, method_name='runYamlMappingTest'):
super().__init__(method_name)
self.data = [
- beam.Row(label='11a', conductor=11, rank=0),
- beam.Row(label='37a', conductor=37, rank=1),
- beam.Row(label='389a', conductor=389, rank=2),
+ beam.Row(
Review Comment:
Might be a bit of an aside, but if we're going to be adding attributes here we should either make them meaningful or arbitrary. (These are actually references to elliptic curves. If we wanted a repeated, nest value, we could provide generators as 2-tuple points [(x=0, y=0)], [(x=0, y=0)], and [(x=0, y=0), (x=1, y=0)] for 11a, 37a, and 389a respectively in the most standard form, but maybe it's easier to just use something else, e.g. col1, if we're trying to explore arbitrary schema types (though this'd be a lot of re-writing).)
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
Review Comment:
Don't mutate the input. You can use dict comprehension here.
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
+ for idx in range(len(row_list)):
+ row_list[idx] = row_to_dict(row_list[idx])
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ PyJsArray,
+ PyJsArrayBuffer,
+ PyJsInt8Array,
+ PyJsUint8Array,
+ PyJsUint8ClampedArray,
+ PyJsInt16Array,
+ PyJsUint16Array,
+ PyJsInt32Array,
+ PyJsUint32Array,
+ PyJsFloat32Array,
+ PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (PyJsNumber, PyJsString, PyJsBoolean)):
+ obj = to_python(obj)
+ elif isinstance(obj, js_array_type):
+ return [_js_object_to_py_object(value) for value in obj.to_list()]
+ elif isinstance(obj, PyJsDate):
+ obj = obj.to_utc_dt()
+ elif isinstance(obj, (PyJsNull, PyJsUndefined)):
+ return None
+ elif isinstance(obj, PyJsError):
+ raise RuntimeError(obj['message'])
+ elif isinstance(obj, PyJsObject):
+ return {
+ key: _js_object_to_py_object(value['value'])
+ for key,
Review Comment:
You can put ()'s around `key, value` to make yapf do the right thing.
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
+ for idx in range(len(row_list)):
+ row_list[idx] = row_to_dict(row_list[idx])
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ PyJsArray,
+ PyJsArrayBuffer,
+ PyJsInt8Array,
+ PyJsUint8Array,
+ PyJsUint8ClampedArray,
+ PyJsInt16Array,
+ PyJsUint16Array,
+ PyJsInt32Array,
+ PyJsUint32Array,
+ PyJsFloat32Array,
+ PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (PyJsNumber, PyJsString, PyJsBoolean)):
+ obj = to_python(obj)
Review Comment:
Same as above. Let's not mix returning directly and mutating obj and returning that.
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
+ for idx in range(len(row_list)):
Review Comment:
Better to use list comprehension `[row_to_dict(x) for x in row]`.
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
+ for idx in range(len(row_list)):
+ row_list[idx] = row_to_dict(row_list[idx])
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ PyJsArray,
+ PyJsArrayBuffer,
+ PyJsInt8Array,
+ PyJsUint8Array,
+ PyJsUint8ClampedArray,
+ PyJsInt16Array,
+ PyJsUint16Array,
+ PyJsInt32Array,
+ PyJsUint32Array,
+ PyJsFloat32Array,
+ PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (PyJsNumber, PyJsString, PyJsBoolean)):
+ obj = to_python(obj)
+ elif isinstance(obj, js_array_type):
+ return [_js_object_to_py_object(value) for value in obj.to_list()]
+ elif isinstance(obj, PyJsDate):
+ obj = obj.to_utc_dt()
+ elif isinstance(obj, (PyJsNull, PyJsUndefined)):
+ return None
+ elif isinstance(obj, PyJsError):
+ raise RuntimeError(obj['message'])
+ elif isinstance(obj, PyJsObject):
+ return {
+ key: _js_object_to_py_object(value['value'])
+ for key,
+ value in obj.own.items()
+ }
+ elif isinstance(obj, JsObjectWrapper):
+ return _js_object_to_py_object(obj._obj)
+
+ return obj
+
+ def _catch_js_errors(func):
+ try:
+ result = func()
+ except JsException as e:
+ result = getattr(e, 'mes')
+ return result
+
if expression:
args = ', '.join(original_fields)
js_func = f'function fn({args}) {{return ({expression})}}'
- js_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
- return lambda __row__: js_callable(*__row__._asdict().values())
+ js_expr_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
+ fn = lambda __row__: lambda: js_expr_callable(
+ *row_to_dict(__row__).values())
elif callable:
js_callable = _CustomJsObjectWrapper(js2py.eval_js(callable))
- return lambda __row__: js_callable(__row__._asdict())
+ fn = lambda __row__: lambda: js_callable(row_to_dict(__row__))
else:
if not path.endswith('.js'):
raise ValueError(f'File "{path}" is not a valid .js file.')
udf_code = FileSystems.open(path).read().decode()
js = js2py.EvalJs()
js.eval(udf_code)
- js_callable = _CustomJsObjectWrapper(getattr(js, name))
- return lambda __row__: js_callable(__row__._asdict())
+ js_file_callable = _CustomJsObjectWrapper(getattr(js, name))
+ fn = lambda __row__: lambda: js_file_callable(row_to_dict(__row__))
+
+ return lambda __row__: dicts_to_rows(
+ _js_object_to_py_object(_catch_js_errors(fn(__row__))))
Review Comment:
I don't thing we want to be catching errors and silently returning them as strings. We should let them propagate (and get processed by the error handling; I think that should just work).
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
Review Comment:
You're never returning row_list.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1375045708
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -27,8 +27,31 @@
from typing import Union
import js2py
+from js2py.base import JsObjectWrapper
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1375066326
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
+ for idx in range(len(row_list)):
+ row_list[idx] = row_to_dict(row_list[idx])
+ return row
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ PyJsArray,
+ PyJsArrayBuffer,
+ PyJsInt8Array,
+ PyJsUint8Array,
+ PyJsUint8ClampedArray,
+ PyJsInt16Array,
+ PyJsUint16Array,
+ PyJsInt32Array,
+ PyJsUint32Array,
+ PyJsFloat32Array,
+ PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (PyJsNumber, PyJsString, PyJsBoolean)):
+ obj = to_python(obj)
+ elif isinstance(obj, js_array_type):
+ return [_js_object_to_py_object(value) for value in obj.to_list()]
+ elif isinstance(obj, PyJsDate):
+ obj = obj.to_utc_dt()
+ elif isinstance(obj, (PyJsNull, PyJsUndefined)):
+ return None
+ elif isinstance(obj, PyJsError):
+ raise RuntimeError(obj['message'])
+ elif isinstance(obj, PyJsObject):
+ return {
+ key: _js_object_to_py_object(value['value'])
+ for key,
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1375043928
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
Review Comment:
I was hoping there was a better way. Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1375055470
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
+ for idx in range(len(row_list)):
Review Comment:
Done.
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +99,89 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def row_to_dict(row):
+ if (str(type(row).__name__).startswith('BeamSchema_') or
+ isinstance(row, Row)):
+ row = row._asdict()
+ if isinstance(row, dict):
+ for key, value in row.items():
+ row[key] = row_to_dict(value)
+ elif not isinstance(row, str) and isinstance(row, Iterable):
+ row_list = list(row)
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "robertwb (via GitHub)" <gi...@apache.org>.
robertwb commented on code in PR #29169:
URL: https://github.com/apache/beam/pull/29169#discussion_r1376941324
##########
sdks/python/apache_beam/yaml/yaml_mapping.py:
##########
@@ -75,28 +79,86 @@ def __setstate__(self, state):
self.__dict__.update(state)
+def py_value_to_js_dict(py_value):
+ if ((isinstance(py_value, tuple) and hasattr(py_value, '_asdict')) or
+ isinstance(py_value, beam.Row)):
+ py_value = py_value._asdict()
+ if isinstance(py_value, dict):
+ return {key: py_value_to_js_dict(value) for key, value in py_value.items()}
+ elif not isinstance(py_value, str) and isinstance(py_value, Iterable):
+ return [py_value_to_js_dict(value) for value in list(py_value)]
+ else:
+ return py_value
+
+
# TODO(yaml) Consider adding optional language version parameter to support
# ECMAScript 5 and 6
def _expand_javascript_mapping_func(
original_fields, expression=None, callable=None, path=None, name=None):
+
+ js_array_type = (
+ base.PyJsArray,
+ base.PyJsArrayBuffer,
+ base.PyJsInt8Array,
+ base.PyJsUint8Array,
+ base.PyJsUint8ClampedArray,
+ base.PyJsInt16Array,
+ base.PyJsUint16Array,
+ base.PyJsInt32Array,
+ base.PyJsUint32Array,
+ base.PyJsFloat32Array,
+ base.PyJsFloat64Array)
+
+ def _js_object_to_py_object(obj):
+ if isinstance(obj, (base.PyJsNumber, base.PyJsString, base.PyJsBoolean)):
+ return base.to_python(obj)
+ elif isinstance(obj, js_array_type):
+ return [_js_object_to_py_object(value) for value in obj.to_list()]
+ elif isinstance(obj, jsdate.PyJsDate):
+ return obj.to_utc_dt()
+ elif isinstance(obj, (base.PyJsNull, base.PyJsUndefined)):
+ return None
+ elif isinstance(obj, base.PyJsError):
+ raise RuntimeError(obj['message'])
+ elif isinstance(obj, base.PyJsObject):
+ return {
+ key: _js_object_to_py_object(value['value'])
+ for (key, value) in obj.own.items()
+ }
+ elif isinstance(obj, base.JsObjectWrapper):
+ return _js_object_to_py_object(obj._obj)
+
+ return obj
+
if expression:
- args = ', '.join(original_fields)
- js_func = f'function fn({args}) {{return ({expression})}}'
- js_callable = _CustomJsObjectWrapper(js2py.eval_js(js_func))
- return lambda __row__: js_callable(*__row__._asdict().values())
+ source = '\n'.join(['function(row) {'] + [
Review Comment:
OK, this is somewhere where you are going to want to do something like __row__ in case they have a field named row.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] fix javascript UDF output in mapping transforms [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #29169:
URL: https://github.com/apache/beam/pull/29169#issuecomment-1783199970
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org