{"id":36493,"date":"2023-03-22T19:26:19","date_gmt":"2023-02-11T05:19:30","guid":{"rendered":"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/"},"modified":"2024-04-29T20:05:52","modified_gmt":"2024-04-29T12:05:52","slug":"apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91","status":"publish","type":"post","link":"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/","title":{"rendered":"Apache Beam\u901f\u67e5\u8868\u3010Python\u3011"},"content":{"rendered":"

\u9996\u5148<\/h1>\n

\u5728\u8fd9\u7bc7\u6587\u7ae0\u4e2d\uff0c\u603b\u7ed3\u4e86Apache Beam Python SDK\u63d0\u4f9b\u7684Transform\u3002\u901a\u8fc7\u4e86\u89e3\u53ef\u4ee5\u65b9\u4fbf\u8c03\u7528\u7684\u5404\u79cdTransform\uff0c\u53ef\u4ee5\u66f4\u5feb\u5730\u5236\u5b9a\u5b9e\u65bd\u7b56\u7565\u3002<\/p>\n

\"undefined\"<\/div>\n

\u9010\u4e2a\u5143\u7d20\u7684\u5904\u7406<\/h1>\n

\u6267\u884cParDo – DoFn<\/h2>\n

\u5bf9\u4e8ePCollection\u4e2d\u7684\u6bcf\u4e2a\u5143\u7d20\uff0c\u8bf7\u6267\u884c\u67d0\u4e9b\u5904\u7406\uff08DoFn\uff09\u3002<\/p>\n

from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> ComputeWordLength<\/span>(<\/span>beam<\/span>.<\/span>DoFn<\/span>):<\/span>\r\n\r\n    def<\/span> __init__<\/span>(<\/span>self<\/span>):<\/span>\r\n        super<\/span>(<\/span>ComputeWordLength<\/span>,<\/span> self<\/span>).<\/span>__init__<\/span>()<\/span>\r\n\r\n    def<\/span> process<\/span>(<\/span>self<\/span>,<\/span> element<\/span>):<\/span>\r\n        yield<\/span> len<\/span>(<\/span>element<\/span>)<\/span>\r\n\r\n\r\nclass<\/span> TestParDo<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_par_do<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>5<\/span>,<\/span> 3<\/span>,<\/span> 7<\/span>,<\/span> 7<\/span>,<\/span> 5<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [<\/span>'Alice'<\/span>,<\/span> 'Bob'<\/span>,<\/span> 'Cameron'<\/span>,<\/span> 'Daniele'<\/span>,<\/span> 'Ellen'<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>ParDo<\/span>(<\/span>ComputeWordLength<\/span>()))<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u8fc7\u6ee4 – \u5bf9\u8981\u7d20\u8fdb\u884c\u7b5b\u9009<\/h2>\n
\u7b5b\u9009 PCollection \u7684\u5143\u7d20\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestFilter<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_filter<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>'A'<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [<\/span>'A'<\/span>,<\/span> 'B'<\/span>,<\/span> 'C'<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Filter<\/span>(<\/span>lambda<\/span> element<\/span>:<\/span> element<\/span>.<\/span>startswith<\/span>(<\/span>'A'<\/span>)))<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u5730\u56fe-\u5bf9\u8981\u7d20\u5e94\u7528\u51fd\u6570<\/h2>\n
\u5bf9 PCollection \u4e2d\u7684\u6bcf\u4e2a\u5143\u7d20\u5e94\u7528\u51fd\u6570\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestMap<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_map<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>5<\/span>,<\/span> 3<\/span>,<\/span> 7<\/span>,<\/span> 7<\/span>,<\/span> 5<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [<\/span>'Alice'<\/span>,<\/span> 'Bob'<\/span>,<\/span> 'Cameron'<\/span>,<\/span> 'Daniele'<\/span>,<\/span> 'Ellen'<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Map<\/span>(<\/span>lambda<\/span> element<\/span>:<\/span> len<\/span>(<\/span>element<\/span>)))<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
FlatMap – \u5bf9\u5143\u7d20\u5e94\u7528\u51fd\u6570\uff08\u53ef\u8fed\u4ee3\uff09<\/h2>\n
\u5bf9 PCollection \u7684\u6bcf\u4e2a\u5143\u7d20\u5e94\u7528\u51fd\u6570\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestFlatMap<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_flat_map<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>5<\/span>,<\/span> 3<\/span>,<\/span> 7<\/span>,<\/span> 7<\/span>,<\/span> 5<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [[<\/span>'Alice'<\/span>,<\/span> 'Bob'<\/span>],<\/span> [<\/span>'Cameron'<\/span>,<\/span> 'Daniele'<\/span>,<\/span> 'Ellen'<\/span>]]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>FlatMap<\/span>(<\/span>lambda<\/span> element<\/span>:<\/span> [<\/span>len<\/span>(<\/span>e<\/span>)<\/span> for<\/span> e<\/span> in<\/span> element<\/span>]))<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
ToString – \u5c06\u8981\u7d20\u8f6c\u6362\u4e3a\u5b57\u7b26\u4e32<\/h2>\n
\u5c06PCollection\u4e2d\u7684\u6bcf\u4e2a\u5143\u7d20\u8f6c\u6362\u4e3a\u5b57\u7b26\u4e32\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestToString<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_to_string_kvs<\/span>(<\/span>self<\/span>):<\/span>\r\n        \"\"\"Key, Value \u3092 , \u533a\u5207\u308a\u306e\u6587\u5b57\u5217\u306b.\"\"\"<\/span>\r\n        expected<\/span> =<\/span> [<\/span>'A,B'<\/span>,<\/span> 'C,D'<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [(<\/span>'A'<\/span>,<\/span> 'B'<\/span>),<\/span> (<\/span>'C'<\/span>,<\/span> 'D'<\/span>)]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>ToString<\/span>.<\/span>Kvs<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n\r\n    def<\/span> test_to_string_element<\/span>(<\/span>self<\/span>):<\/span>\r\n        \"\"\"\u5404\u8981\u7d20\u3092\u6587\u5b57\u5217\u306b.\"\"\"<\/span>\r\n        expected<\/span> =<\/span> [<\/span>\"A\"<\/span>,<\/span> \"['A', 'B']\"<\/span>,<\/span> \"['C', 'D', 'E']\"<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [<\/span>'A'<\/span>,<\/span> [<\/span>'A'<\/span>,<\/span> 'B'<\/span>],<\/span> [<\/span>'C'<\/span>,<\/span> 'D'<\/span>,<\/span> 'E'<\/span>]]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>ToString<\/span>.<\/span>Element<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n\r\n    def<\/span> test_to_string_iterables<\/span>(<\/span>self<\/span>):<\/span>\r\n        \"\"\"\u30a4\u30c6\u30e9\u30d6\u30eb\u306a\u30aa\u30d6\u30b8\u30a7\u30af\u30c8\u3092\u6587\u5b57\u5217\u306b.\"\"\"<\/span>\r\n        expected<\/span> =<\/span> [<\/span>'A,B'<\/span>,<\/span> 'C,D,E'<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [[<\/span>'A'<\/span>,<\/span> 'B'<\/span>],<\/span> [<\/span>'C'<\/span>,<\/span> 'D'<\/span>,<\/span> 'E'<\/span>]]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>ToString<\/span>.<\/span>Iterables<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u4ece\u8981\u7d20\u4e2d\u63d0\u53d6\u952e<\/h2>\n
\u4ece PCollection \u7684\u6bcf\u4e2a\u5143\u7d20\uff08\u952e\u503c\u5bf9\uff09\u4e2d\u62bd\u53d6\u952e\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestKeys<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_keys<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>0<\/span>,<\/span> 1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>,<\/span> 4<\/span>,<\/span> 5<\/span>,<\/span> 6<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [(<\/span>0<\/span>,<\/span> 'Sunday'<\/span>),<\/span> (<\/span>1<\/span>,<\/span> 'Monday'<\/span>),<\/span> (<\/span>2<\/span>,<\/span> 'Tuesday'<\/span>),<\/span> (<\/span>3<\/span>,<\/span> 'Wednesday'<\/span>),<\/span>\r\n                  (<\/span>4<\/span>,<\/span> 'Thursday'<\/span>),<\/span> (<\/span>5<\/span>,<\/span> 'Friday'<\/span>),<\/span> (<\/span>6<\/span>,<\/span> 'Saturday'<\/span>)]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Keys<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u4ece\u8981\u7d20\u4e2d\u63d0\u53d6\u4ef7\u503c<\/h2>\n
\u4ece PCollection \u7684\u6bcf\u4e2a\u5143\u7d20\uff08\u952e\u548c\u503c\u5bf9\uff09\u4e2d\u63d0\u53d6\u503c\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestValues<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_values<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>'Sunday'<\/span>,<\/span> 'Monday'<\/span>,<\/span> 'Tuesday'<\/span>,<\/span> 'Wednesday'<\/span>,<\/span> 'Thursday'<\/span>,<\/span> 'Friday'<\/span>,<\/span> 'Saturday'<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [(<\/span>0<\/span>,<\/span> 'Sunday'<\/span>),<\/span> (<\/span>1<\/span>,<\/span> 'Monday'<\/span>),<\/span> (<\/span>2<\/span>,<\/span> 'Tuesday'<\/span>),<\/span> (<\/span>3<\/span>,<\/span> 'Wednesday'<\/span>),<\/span>\r\n                  (<\/span>4<\/span>,<\/span> 'Thursday'<\/span>),<\/span> (<\/span>5<\/span>,<\/span> 'Friday'<\/span>),<\/span> (<\/span>6<\/span>,<\/span> 'Saturday'<\/span>)]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Values<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
KvSwap – \u4ea4\u6362\u5143\u7d20\u7684\u952e\u548c\u503c<\/h2>\n
\u5c06 PCollection \u4e2d\u7684\u6bcf\u4e2a\u5143\u7d20\uff08\u952e\u503c\u5bf9\uff09\u7684\u952e\u548c\u503c\u8fdb\u884c\u4e92\u6362\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestKvSwap<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_kv_swap<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [(<\/span>'Friday'<\/span>,<\/span> 5<\/span>),<\/span> (<\/span>'Monday'<\/span>,<\/span> 1<\/span>),<\/span> (<\/span>'Saturday'<\/span>,<\/span> 6<\/span>),<\/span> (<\/span>'Sunday'<\/span>,<\/span> 0<\/span>),<\/span>\r\n                    (<\/span>'Thursday'<\/span>,<\/span> 4<\/span>),<\/span> (<\/span>'Tuesday'<\/span>,<\/span> 2<\/span>),<\/span> (<\/span>'Wednesday'<\/span>,<\/span> 3<\/span>)]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [(<\/span>0<\/span>,<\/span> 'Sunday'<\/span>),<\/span> (<\/span>1<\/span>,<\/span> 'Monday'<\/span>),<\/span> (<\/span>2<\/span>,<\/span> 'Tuesday'<\/span>),<\/span> (<\/span>3<\/span>,<\/span> 'Wednesday'<\/span>),<\/span>\r\n                  (<\/span>4<\/span>,<\/span> 'Thursday'<\/span>),<\/span> (<\/span>5<\/span>,<\/span> 'Friday'<\/span>),<\/span> (<\/span>6<\/span>,<\/span> 'Saturday'<\/span>)]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>KvSwap<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u805a\u5408\u5904\u7406<\/h1>\n
\u6309\u7167\u952e\uff08Key\uff09\u5bf9\u8981\u7d20\u8fdb\u884c\u5206\u7ec4\u805a\u5408 – GroupByKey<\/h2>\n
\u6309\u7167 Key \u5c06 PCollection \u4e2d\u7684\u5143\u7d20\uff08\u5305\u542b\u952e\u548c\u503c\u7684\u5bf9\uff09\u8fdb\u884c\u805a\u5408\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestGroupByKey<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_group_by_key<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [(<\/span>'cat'<\/span>,<\/span> [<\/span>'tama'<\/span>,<\/span> 'mike'<\/span>]),<\/span> (<\/span>'dog'<\/span>,<\/span> [<\/span>'pochi'<\/span>])]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [(<\/span>'cat'<\/span>,<\/span> 'tama'<\/span>),<\/span> (<\/span>'cat'<\/span>,<\/span> 'mike'<\/span>),<\/span> (<\/span>'dog'<\/span>,<\/span> 'pochi'<\/span>)]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>GroupByKey<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
CoGroupByKey – \u4f7f\u7528 Key \u5c06\u8981\u7d20\u805a\u5408\u5728\u4e00\u8d77\uff08\u591a\u4e2a PCollection\uff09<\/h2>\n
\u6309\u7167\u952e\uff08Key\uff09\u7684\u65b9\u5f0f\u5bf9\u591a\u4e2a PCollection \u7684\u5143\u7d20\uff08\u952e\u503c\u5bf9\uff09\u8fdb\u884c\u805a\u5408\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestCoGroupByKey<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_co_group_by_key<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>\r\n            (<\/span>'amy'<\/span>,<\/span> ([<\/span>'amy@example.com'<\/span>],<\/span> [<\/span>'111-222-3333'<\/span>,<\/span> '333-444-5555'<\/span>])),<\/span>\r\n            (<\/span>'julia'<\/span>,<\/span> ([<\/span>'julia@example.com'<\/span>],<\/span> []))<\/span>\r\n        ]<\/span>\r\n\r\n        inputs1<\/span> =<\/span> [(<\/span>'amy'<\/span>,<\/span> 'amy@example.com'<\/span>),<\/span> (<\/span>'julia'<\/span>,<\/span> 'julia@example.com'<\/span>)]<\/span>\r\n        inputs2<\/span> =<\/span> [(<\/span>'amy'<\/span>,<\/span> '111-222-3333'<\/span>),<\/span> (<\/span>'amy'<\/span>,<\/span> '333-444-5555'<\/span>)]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            pcol1<\/span> =<\/span> p<\/span> |<\/span> 'create pcol1'<\/span> &<\/span>gt<\/span>;<\/span>&<\/span>gt<\/span>;<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs1<\/span>)<\/span>\r\n            pcol2<\/span> =<\/span> p<\/span> |<\/span> 'create pcol2'<\/span> &<\/span>gt<\/span>;<\/span>&<\/span>gt<\/span>;<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs2<\/span>)<\/span>\r\n\r\n            actual<\/span> =<\/span> ((<\/span>pcol1<\/span>,<\/span> pcol2<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>CoGroupByKey<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u5168\u7403\u5408\u5e76 – \u8981\u7d20\u7684\u7ed3\u5408<\/h2>\n
\u5c06 PCollection \u4e2d\u7684\u6240\u6709\u5143\u7d20\u5408\u5e76\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestCombineGlobally<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_combine_globally<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>55<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [<\/span>1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>,<\/span> 4<\/span>,<\/span> 5<\/span>,<\/span> 6<\/span>,<\/span> 7<\/span>,<\/span> 8<\/span>,<\/span> 9<\/span>,<\/span> 10<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>CombineGlobally<\/span>(<\/span>sum<\/span>))<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u5c06\u8981\u7d20\u5b58\u50a8\u5230\u4e00\u4e2a\u5217\u8868\u4e2d<\/h2>\n
\u5c06 PCollection \u7684\u6240\u6709\u5143\u7d20\u5b58\u50a8\u5728\u4e00\u4e2a\u5217\u8868\u4e2d\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestToList<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_to_list<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [[<\/span>1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>,<\/span> 4<\/span>,<\/span> 5<\/span>,<\/span> 6<\/span>,<\/span> 7<\/span>,<\/span> 8<\/span>,<\/span> 9<\/span>,<\/span> 10<\/span>]]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [<\/span>1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>,<\/span> 4<\/span>,<\/span> 5<\/span>,<\/span> 6<\/span>,<\/span> 7<\/span>,<\/span> 8<\/span>,<\/span> 9<\/span>,<\/span> 10<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>combiners<\/span>.<\/span>ToList<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u5c06\u8981\u7d20\u5b58\u50a8\u4e3a\u4e00\u4e2a\u5b57\u5178\u7c7b\u578b\uff0c\u4f7f\u7528ToDict(\uff09\u51fd\u6570\u3002<\/h2>\n
\u5c06 PCollection \u4e2d\u7684\u6240\u6709\u5143\u7d20\uff08\u952e\u548c\u503c\u5bf9\uff09\u5b58\u50a8\u5230\u4e00\u4e2a\u5b57\u5178\u7c7b\u578b\u4e2d\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestToDict<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_to_dict<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [{<\/span>'A'<\/span>:<\/span> 2<\/span>,<\/span> 'B'<\/span>:<\/span> 1<\/span>}]<\/span>  # Key \u304c\u88ab\u308b\u5834\u5408\u306f\u3069\u3061\u3089\u304b\u4e00\u65b9\u306e Value \u304c\u9078\u629e\u3055\u308c\u308b\r\n<\/span>\r\n        inputs<\/span> =<\/span> [(<\/span>'A'<\/span>,<\/span> 1<\/span>),<\/span> (<\/span>'A'<\/span>,<\/span> 2<\/span>),<\/span> (<\/span>'B'<\/span>,<\/span> 1<\/span>)]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>combiners<\/span>.<\/span>ToDict<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u8ba1\u7b97 – \u7edf\u8ba1\u8981\u7d20\u7684\u6570\u91cf<\/h2>\n
\u8ba1\u7b97 PCollection \u7684\u5143\u7d20\u6570\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestCount<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_count<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>10<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [<\/span>1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>,<\/span> 4<\/span>,<\/span> 5<\/span>,<\/span> 6<\/span>,<\/span> 7<\/span>,<\/span> 8<\/span>,<\/span> 9<\/span>,<\/span> 10<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>combiners<\/span>.<\/span>Count<\/span>.<\/span>Globally<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u533a\u522b\u660e\u663e – \u9664\u53bb\u91cd\u590d\u8981\u7d20<\/h2>\n
\u4ece PCollection \u7684\u5143\u7d20\u4e2d\u6392\u9664\u91cd\u590d\u9879\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestDistinct<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_distinct<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [<\/span>1<\/span>,<\/span> 1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Distinct<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u8ba1\u7b97\u8981\u7d20\u7684\u5e73\u5747\u503c<\/h2>\n
\u8ba1\u7b97PCollection\u4e2d\u6240\u6709\u5143\u7d20\u7684\u5e73\u5747\u503c\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestMean<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_mean<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>5.5<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [<\/span>1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>,<\/span> 4<\/span>,<\/span> 5<\/span>,<\/span> 6<\/span>,<\/span> 7<\/span>,<\/span> 8<\/span>,<\/span> 9<\/span>,<\/span> 10<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>combiners<\/span>.<\/span>Mean<\/span>.<\/span>Globally<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u6837\u672c – \u4ece\u8981\u7d20\u4e2d\u968f\u673a\u62bd\u53d6<\/h2>\n
\u4ece PCollection \u4e2d\u968f\u673a\u62bd\u53d6\u82e5\u5e72\u9879\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestSample<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_sample<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [[<\/span>2<\/span>,<\/span> 8<\/span>,<\/span> 6<\/span>]]<\/span>  # \u671f\u5f85\u5024\u306f\u6bce\u56de\u30e9\u30f3\u30c0\u30e0\u306a\u5024\u306b\u306a\u308b\r\n<\/span>\r\n        inputs<\/span> =<\/span> [<\/span>1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>,<\/span> 4<\/span>,<\/span> 5<\/span>,<\/span> 6<\/span>,<\/span> 7<\/span>,<\/span> 8<\/span>,<\/span> 9<\/span>,<\/span> 10<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>combiners<\/span>.<\/span>Sample<\/span>.<\/span>FixedSizeGlobally<\/span>(<\/span>3<\/span>))<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u9876\u90e8-\u4ece\u8981\u7d20\u4e2d\u63d0\u53d6\u6700\u5927\uff08\u6216\u6700\u5c0f\uff09\u503c<\/h2>\n
\u4ecePCollection\u7684\u6240\u6709\u5143\u7d20\u4e2d\u63d0\u53d6\u51fa\u6700\u5927\uff08\u6216\u6700\u5c0f\uff09\u7684\u6570\u4e2a\u5143\u7d20\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestTop<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_top_largest<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [[<\/span>10<\/span>,<\/span> 9<\/span>,<\/span> 8<\/span>]]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [<\/span>1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>,<\/span> 4<\/span>,<\/span> 5<\/span>,<\/span> 6<\/span>,<\/span> 7<\/span>,<\/span> 8<\/span>,<\/span> 9<\/span>,<\/span> 10<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>combiners<\/span>.<\/span>Top<\/span>.<\/span>Largest<\/span>(<\/span>3<\/span>))<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n\r\n    def<\/span> test_top_smallest<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [[<\/span>1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>]]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [<\/span>1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>,<\/span> 4<\/span>,<\/span> 5<\/span>,<\/span> 6<\/span>,<\/span> 7<\/span>,<\/span> 8<\/span>,<\/span> 9<\/span>,<\/span> 10<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>combiners<\/span>.<\/span>Top<\/span>.<\/span>Smallest<\/span>(<\/span>3<\/span>))<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u5176\u4ed6\u5904\u7406 | Others<\/h1>\n
\u5408\u5e76 – PCollection \u7684\u6574\u7406<\/h2>\n
\u5c06\u591a\u4e2a PCollection \u5408\u5e76\u6210\u4e00\u4e2a PCollection\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestFlatten<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_flatten<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>,<\/span> 4<\/span>,<\/span> 5<\/span>,<\/span> 6<\/span>,<\/span> 7<\/span>,<\/span> 8<\/span>,<\/span> 9<\/span>,<\/span> 10<\/span>]<\/span>\r\n\r\n        inputs1<\/span> =<\/span> [<\/span>1<\/span>,<\/span> 2<\/span>,<\/span> 3<\/span>,<\/span> 4<\/span>,<\/span> 5<\/span>]<\/span>\r\n        inputs2<\/span> =<\/span> [<\/span>6<\/span>,<\/span> 7<\/span>,<\/span> 8<\/span>,<\/span> 9<\/span>,<\/span> 10<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            pcol1<\/span> =<\/span> p<\/span> |<\/span> 'create pcol1'<\/span> >><\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs1<\/span>)<\/span>\r\n            pcol2<\/span> =<\/span> p<\/span> |<\/span> 'create pcol2'<\/span> >><\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs2<\/span>)<\/span>\r\n\r\n            actual<\/span> =<\/span> (<\/span>pcol1<\/span>,<\/span> pcol2<\/span>)<\/span> |<\/span> beam<\/span>.<\/span>Flatten<\/span>()<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u8c03\u6574 – \u8981\u7d20\u7684\u91cd\u65b0\u5206\u914d<\/h2>\n
\u5728\u5de5\u4f5c\u673a\u4e4b\u95f4\u91cd\u65b0\u5206\u914d PCollection \u7684\u5143\u7d20\u3002<\/p>\n
from<\/span> unittest<\/span> import<\/span> TestCase<\/span>\r\n\r\nimport<\/span> apache_beam<\/span> as<\/span> beam<\/span>\r\nfrom<\/span> apache_beam.testing.test_pipeline<\/span> import<\/span> TestPipeline<\/span>\r\nfrom<\/span> apache_beam.testing.util<\/span> import<\/span> assert_that<\/span>,<\/span> equal_to<\/span>\r\n\r\n\r\nclass<\/span> TestReshuffle<\/span>(<\/span>TestCase<\/span>):<\/span>\r\n\r\n    def<\/span> test_reshuffle<\/span>(<\/span>self<\/span>):<\/span>\r\n        expected<\/span> =<\/span> [<\/span>'A'<\/span>,<\/span> 'B'<\/span>,<\/span> 'C'<\/span>]<\/span>\r\n\r\n        inputs<\/span> =<\/span> [<\/span>'A'<\/span>,<\/span> 'B'<\/span>,<\/span> 'C'<\/span>]<\/span>\r\n\r\n        with<\/span> TestPipeline<\/span>()<\/span> as<\/span> p<\/span>:<\/span>\r\n            actual<\/span> =<\/span> (<\/span>p<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Create<\/span>(<\/span>inputs<\/span>)<\/span>\r\n                      |<\/span> beam<\/span>.<\/span>Reshuffle<\/span>())<\/span>\r\n\r\n            assert_that<\/span>(<\/span>actual<\/span>,<\/span> equal_to<\/span>(<\/span>expected<\/span>))<\/span>\r\n<\/code><\/pre>\n
\u603b\u7ed3<\/h1>\n
\u5728Apache Beam\u7684Python SDK\u4e2d\uff0c\u63d0\u4f9b\u4e86\u4e30\u5bcc\u7684Transform\u9009\u9879\uff08\u76f8\u8f83\u4e8eJava\u6765\u8bf4\u8f83\u5c11\uff09\u3002\u6211\u4eec\u5c06\u968f\u65f6\u66f4\u65b0\u4ee5\u63d0\u4f9b\u65b0\u7684\u529f\u80fd\u3002<\/p>\n
\u5e0c\u671b\u4f60\u4eec\u80fd\u5728\u9700\u8981\u5feb\u901f\u56de\u60f3\u8d77 Apache Beam \u7684 Transform \u65f6\u53c2\u8003\u8fd9\u4e2a\u4fe1\u606f\uff01<\/p>\n
\u8bf7\u53c2\u8003\u4ee5\u4e0b\u7f51\u5740<\/h1>\n
    Python transform catalog<\/ul>\n","protected":false},"excerpt":{"rendered":"
    \u9996\u5148 \u5728\u8fd9\u7bc7\u6587\u7ae0\u4e2d\uff0c\u603b\u7ed3\u4e86Apache Beam Python SDK\u63d0\u4f9b\u7684Transform\u3002\u901a\u8fc7\u4e86\u89e3\u53ef\u4ee5\u65b9 […]<\/p>\n","protected":false},"author":5,"featured_media":0,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-36493","post","type-post","status-publish","format-standard","hentry","category-uncategorized"],"yoast_head":"\nApache Beam\u901f\u67e5\u8868\u3010Python\u3011 - Blog - Silicon Cloud<\/title>\n<meta name=\"robots\" content=\"index, follow, max-snippet:-1, max-image-preview:large, max-video-preview:-1\" \/>\n<link rel=\"canonical\" href=\"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam\u901f\u67e5\u8868\u3010python\u3011\/\" \/>\n<meta property=\"og:locale\" content=\"zh_CN\" \/>\n<meta property=\"og:type\" content=\"article\" \/>\n<meta property=\"og:title\" content=\"Apache Beam\u901f\u67e5\u8868\u3010Python\u3011\" \/>\n<meta property=\"og:description\" content=\"\u9996\u5148 \u5728\u8fd9\u7bc7\u6587\u7ae0\u4e2d\uff0c\u603b\u7ed3\u4e86Apache Beam Python SDK\u63d0\u4f9b\u7684Transform\u3002\u901a\u8fc7\u4e86\u89e3\u53ef\u4ee5\u65b9 […]\" \/>\n<meta property=\"og:url\" content=\"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam\u901f\u67e5\u8868\u3010python\u3011\/\" \/>\n<meta property=\"og:site_name\" content=\"Blog - Silicon Cloud\" \/>\n<meta property=\"article:published_time\" content=\"2023-02-11T05:19:30+00:00\" \/>\n<meta property=\"article:modified_time\" content=\"2024-04-29T12:05:52+00:00\" \/>\n<meta property=\"og:image\" content=\"https:\/\/cdn.silicloud.com\/blog-img\/blog\/img\/657d295737434c4406c4373c\/2-0.png\" \/>\n<meta name=\"author\" content=\"\u6e05, \u5b87\" \/>\n<meta name=\"twitter:card\" content=\"summary_large_image\" \/>\n<meta name=\"twitter:label1\" content=\"\u4f5c\u8005\" \/>\n\t<meta name=\"twitter:data1\" content=\"\u6e05, \u5b87\" \/>\n\t<meta name=\"twitter:label2\" content=\"\u9884\u8ba1\u9605\u8bfb\u65f6\u95f4\" \/>\n\t<meta name=\"twitter:data2\" content=\"7 \u5206\" \/>\n<script type=\"application\/ld+json\" class=\"yoast-schema-graph\">{\"@context\":\"https:\/\/schema.org\",\"@graph\":[{\"@type\":\"WebPage\",\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/\",\"url\":\"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/\",\"name\":\"Apache Beam\u901f\u67e5\u8868\u3010Python\u3011 - Blog - Silicon Cloud\",\"isPartOf\":{\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/#website\"},\"datePublished\":\"2023-02-11T05:19:30+00:00\",\"dateModified\":\"2024-04-29T12:05:52+00:00\",\"author\":{\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/#\/schema\/person\/1a6ecd3d914d22a5ac32791ffc1fbd8e\"},\"breadcrumb\":{\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/#breadcrumb\"},\"inLanguage\":\"zh-Hans\",\"potentialAction\":[{\"@type\":\"ReadAction\",\"target\":[\"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/\"]}]},{\"@type\":\"BreadcrumbList\",\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/#breadcrumb\",\"itemListElement\":[{\"@type\":\"ListItem\",\"position\":1,\"name\":\"\u9996\u9875\",\"item\":\"https:\/\/www.silicloud.com\/zh\/blog\/\"},{\"@type\":\"ListItem\",\"position\":2,\"name\":\"Apache Beam\u901f\u67e5\u8868\u3010Python\u3011\"}]},{\"@type\":\"WebSite\",\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/#website\",\"url\":\"https:\/\/www.silicloud.com\/zh\/blog\/\",\"name\":\"Blog - Silicon Cloud\",\"description\":\"\",\"inLanguage\":\"zh-Hans\"},{\"@type\":\"Person\",\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/#\/schema\/person\/1a6ecd3d914d22a5ac32791ffc1fbd8e\",\"name\":\"\u6e05, \u5b87\",\"image\":{\"@type\":\"ImageObject\",\"inLanguage\":\"zh-Hans\",\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/#\/schema\/person\/image\/\",\"url\":\"https:\/\/secure.gravatar.com\/avatar\/4b2016c18459a605fc469c7566608f5686491baa112d0871ee613f61b7210565?s=96&d=mm&r=g\",\"contentUrl\":\"https:\/\/secure.gravatar.com\/avatar\/4b2016c18459a605fc469c7566608f5686491baa112d0871ee613f61b7210565?s=96&d=mm&r=g\",\"caption\":\"\u6e05, \u5b87\"},\"url\":\"https:\/\/www.silicloud.com\/zh\/blog\/author\/qingyu\/\"},{\"@type\":\"ImageObject\",\"inLanguage\":\"zh-Hans\",\"@id\":\"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/#local-main-organization-logo\",\"url\":\"\",\"contentUrl\":\"\",\"caption\":\"Blog - Silicon Cloud\"}]}<\/script>\n<!-- \/ Yoast SEO Premium plugin. -->","yoast_head_json":{"title":"Apache Beam\u901f\u67e5\u8868\u3010Python\u3011 - Blog - Silicon Cloud","robots":{"index":"index","follow":"follow","max-snippet":"max-snippet:-1","max-image-preview":"max-image-preview:large","max-video-preview":"max-video-preview:-1"},"canonical":"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam\u901f\u67e5\u8868\u3010python\u3011\/","og_locale":"zh_CN","og_type":"article","og_title":"Apache Beam\u901f\u67e5\u8868\u3010Python\u3011","og_description":"\u9996\u5148 \u5728\u8fd9\u7bc7\u6587\u7ae0\u4e2d\uff0c\u603b\u7ed3\u4e86Apache Beam Python SDK\u63d0\u4f9b\u7684Transform\u3002\u901a\u8fc7\u4e86\u89e3\u53ef\u4ee5\u65b9 […]","og_url":"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam\u901f\u67e5\u8868\u3010python\u3011\/","og_site_name":"Blog - Silicon Cloud","article_published_time":"2023-02-11T05:19:30+00:00","article_modified_time":"2024-04-29T12:05:52+00:00","og_image":[{"url":"https:\/\/cdn.silicloud.com\/blog-img\/blog\/img\/657d295737434c4406c4373c\/2-0.png"}],"author":"\u6e05, \u5b87","twitter_card":"summary_large_image","twitter_misc":{"\u4f5c\u8005":"\u6e05, \u5b87","\u9884\u8ba1\u9605\u8bfb\u65f6\u95f4":"7 \u5206"},"schema":{"@context":"https:\/\/schema.org","@graph":[{"@type":"WebPage","@id":"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/","url":"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/","name":"Apache Beam\u901f\u67e5\u8868\u3010Python\u3011 - Blog - Silicon Cloud","isPartOf":{"@id":"https:\/\/www.silicloud.com\/zh\/blog\/#website"},"datePublished":"2023-02-11T05:19:30+00:00","dateModified":"2024-04-29T12:05:52+00:00","author":{"@id":"https:\/\/www.silicloud.com\/zh\/blog\/#\/schema\/person\/1a6ecd3d914d22a5ac32791ffc1fbd8e"},"breadcrumb":{"@id":"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/#breadcrumb"},"inLanguage":"zh-Hans","potentialAction":[{"@type":"ReadAction","target":["https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/"]}]},{"@type":"BreadcrumbList","@id":"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/#breadcrumb","itemListElement":[{"@type":"ListItem","position":1,"name":"\u9996\u9875","item":"https:\/\/www.silicloud.com\/zh\/blog\/"},{"@type":"ListItem","position":2,"name":"Apache Beam\u901f\u67e5\u8868\u3010Python\u3011"}]},{"@type":"WebSite","@id":"https:\/\/www.silicloud.com\/zh\/blog\/#website","url":"https:\/\/www.silicloud.com\/zh\/blog\/","name":"Blog - Silicon Cloud","description":"","inLanguage":"zh-Hans"},{"@type":"Person","@id":"https:\/\/www.silicloud.com\/zh\/blog\/#\/schema\/person\/1a6ecd3d914d22a5ac32791ffc1fbd8e","name":"\u6e05, \u5b87","image":{"@type":"ImageObject","inLanguage":"zh-Hans","@id":"https:\/\/www.silicloud.com\/zh\/blog\/#\/schema\/person\/image\/","url":"https:\/\/secure.gravatar.com\/avatar\/4b2016c18459a605fc469c7566608f5686491baa112d0871ee613f61b7210565?s=96&d=mm&r=g","contentUrl":"https:\/\/secure.gravatar.com\/avatar\/4b2016c18459a605fc469c7566608f5686491baa112d0871ee613f61b7210565?s=96&d=mm&r=g","caption":"\u6e05, \u5b87"},"url":"https:\/\/www.silicloud.com\/zh\/blog\/author\/qingyu\/"},{"@type":"ImageObject","inLanguage":"zh-Hans","@id":"https:\/\/www.silicloud.com\/zh\/blog\/apache-beam%e9%80%9f%e6%9f%a5%e8%a1%a8%e3%80%90python%e3%80%91\/#local-main-organization-logo","url":"","contentUrl":"","caption":"Blog - Silicon Cloud"}]}},"_links":{"self":[{"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/posts\/36493","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/users\/5"}],"replies":[{"embeddable":true,"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/comments?post=36493"}],"version-history":[{"count":2,"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/posts\/36493\/revisions"}],"predecessor-version":[{"id":87364,"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/posts\/36493\/revisions\/87364"}],"wp:attachment":[{"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/media?parent=36493"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/categories?post=36493"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.silicloud.com\/zh\/blog\/wp-json\/wp\/v2\/tags?post=36493"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}