[FLINK-4795] [py] Fix CsvStringify for nested tuples
Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/af96c077 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/af96c077 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/af96c077 Branch: refs/heads/master Commit: af96c0776ea1a48838f1d452efb42c03caa64251 Parents: 6f0faf9 Author: zentol <[email protected]> Authored: Wed Oct 12 12:04:58 2016 +0200 Committer: zentol <[email protected]> Committed: Fri Oct 21 11:03:03 2016 +0200 ---------------------------------------------------------------------- .../python/org/apache/flink/python/api/flink/plan/DataSet.py | 2 +- .../src/test/python/org/apache/flink/python/api/test_main.py | 4 ++++ 2 files changed, 5 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/af96c077/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/plan/DataSet.py ---------------------------------------------------------------------- diff --git a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/plan/DataSet.py b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/plan/DataSet.py index 06557ca..488bcb8 100644 --- a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/plan/DataSet.py +++ b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/plan/DataSet.py @@ -51,7 +51,7 @@ class CsvStringify(MapFunction): def _map(self, value): if isinstance(value, (tuple, list)): - return "(" + b", ".join([self.map(x) for x in value]) + ")" + return "(" + b", ".join([self._map(x) for x in value]) + ")" else: return str(value) http://git-wip-us.apache.org/repos/asf/flink/blob/af96c077/flink-libraries/flink-python/src/test/python/org/apache/flink/python/api/test_main.py ---------------------------------------------------------------------- diff --git a/flink-libraries/flink-python/src/test/python/org/apache/flink/python/api/test_main.py b/flink-libraries/flink-python/src/test/python/org/apache/flink/python/api/test_main.py index 223ff68..29eb36b 100644 --- a/flink-libraries/flink-python/src/test/python/org/apache/flink/python/api/test_main.py +++ b/flink-libraries/flink-python/src/test/python/org/apache/flink/python/api/test_main.py @@ -44,6 +44,8 @@ if __name__ == "__main__": d7 = env.generate_sequence(0, 999) + d8 = env.from_elements((1, (2, 3)), (4, (5, 6))) + #Generate Sequence Source d7.map(Id()).map_partition(Verify(range(1000), "Sequence")).output() @@ -58,6 +60,8 @@ if __name__ == "__main__": csv_data.write_csv("/tmp/flink/result1", line_delimiter="\n", field_delimiter="|", write_mode=WriteMode.OVERWRITE) + d8.write_csv("/tmp/flink/result2", line_delimiter="\n", field_delimiter="|", write_mode=WriteMode.OVERWRITE) + #Text Source/Sink text_data = env.read_text("src/test/python/org/apache/flink/python/api/data_text")
