This is an automated email from the ASF dual-hosted git repository. lindong pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink-ml.git
commit c8fef6054f83c1dc1dc188606b61bd5a92c60c57 Author: yunfengzhou-hub <[email protected]> AuthorDate: Thu May 26 20:50:20 2022 +0800 [FLINK-27798] Fix python test for Flink 1.15 migration. This closes #104. --- .../pyflink/ml/core/tests/test_pipeline.py | 23 ++++++++++++---------- .../ml/lib/feature/tests/test_vectorassembler.py | 22 +++++++++++++++------ 2 files changed, 29 insertions(+), 16 deletions(-) diff --git a/flink-ml-python/pyflink/ml/core/tests/test_pipeline.py b/flink-ml-python/pyflink/ml/core/tests/test_pipeline.py index ac5e662..754c9ee 100644 --- a/flink-ml-python/pyflink/ml/core/tests/test_pipeline.py +++ b/flink-ml-python/pyflink/ml/core/tests/test_pipeline.py @@ -18,8 +18,6 @@ import os from typing import Dict, Any, List -import pandas as pd -from pandas._testing import assert_frame_equal from pyflink.table import Table, StreamTableEnvironment from pyflink.ml.core.api import Model @@ -39,8 +37,9 @@ class PipelineTest(PyFlinkMLTestCase): model = PipelineModel([model_a, model_b, model_c]) output_table = model.transform(input_table)[0] - assert_frame_equal(output_table.to_pandas(), - pd.DataFrame([[31], [32], [33]], columns=['a'])) + predicted_results = [result[0] for result in + self.t_env.to_data_stream(output_table).execute_and_collect()] + self.assertEqual(predicted_results, [31, 32, 33]) # Saves and loads the PipelineModel. path = os.path.join(self.temp_dir, "test_pipeline_model") @@ -48,8 +47,9 @@ class PipelineTest(PyFlinkMLTestCase): loaded_model = PipelineModel.load(self.t_env, path) output_table2 = loaded_model.transform(input_table)[0] - assert_frame_equal(output_table2.to_pandas(), - pd.DataFrame([[31], [32], [33]], columns=['a'])) + predicted_results = [result[0] for result in + self.t_env.to_data_stream(output_table2).execute_and_collect()] + self.assertEqual(predicted_results, [31, 32, 33]) def test_pipeline(self): input_table = self.t_env.from_elements([(1,), (2,), (3,)], ['a']) @@ -60,8 +60,9 @@ class PipelineTest(PyFlinkMLTestCase): model = estimator.fit(input_table) output_table = model.transform(input_table)[0] - assert_frame_equal(output_table.to_pandas(), - pd.DataFrame([[21], [22], [23]], columns=['a'])) + predicted_results = [result[0] for result in + self.t_env.to_data_stream(output_table).execute_and_collect()] + self.assertEqual(predicted_results, [21, 22, 23]) # Saves and loads the PipelineModel. path = os.path.join(self.temp_dir, "test_pipeline") @@ -70,8 +71,10 @@ class PipelineTest(PyFlinkMLTestCase): model = loaded_estimator.fit(input_table) output_table = model.transform(input_table)[0] - assert_frame_equal(output_table.to_pandas(), - pd.DataFrame([[21], [22], [23]], columns=['a'])) + + predicted_results = [result[0] for result in + self.t_env.to_data_stream(output_table).execute_and_collect()] + self.assertEqual(predicted_results, [21, 22, 23]) class Add10Model(Model): diff --git a/flink-ml-python/pyflink/ml/lib/feature/tests/test_vectorassembler.py b/flink-ml-python/pyflink/ml/lib/feature/tests/test_vectorassembler.py index c72e6dd..5362fcf 100644 --- a/flink-ml-python/pyflink/ml/lib/feature/tests/test_vectorassembler.py +++ b/flink-ml-python/pyflink/ml/lib/feature/tests/test_vectorassembler.py @@ -15,6 +15,7 @@ # See the License for the specific language governing permissions and # limitations under the License. ################################################################################ +import os from pyflink.common import Types @@ -26,6 +27,7 @@ from pyflink.ml.tests.test_utils import PyFlinkMLTestCase class VectorAssemblerTest(PyFlinkMLTestCase): def setUp(self): super(VectorAssemblerTest, self).setUp() + # TODO: Add test for handling invalid values after FLINK-27797 is resolved. self.input_data_table = self.t_env.from_data_stream( self.env.from_collection([ (0, @@ -37,7 +39,6 @@ class VectorAssemblerTest(PyFlinkMLTestCase): 1.0, Vectors.sparse(5, [1, 2, 3, 4], [1.0, 2.0, 3.0, 4.0])), - (2, None, None, None), ], type_info=Types.ROW_NAMED( ['id', 'vec', 'num', 'sparse_vec'], @@ -60,13 +61,22 @@ class VectorAssemblerTest(PyFlinkMLTestCase): self.assertEqual('skip', vector_assembler.handle_invalid) self.assertEqual('assembled_vec', vector_assembler.output_col) - def test_keep_invalid(self): + def test_save_load_transform(self): vector_assembler = VectorAssembler() \ .set_input_cols('vec', 'num', 'sparse_vec') \ .set_output_col('assembled_vec') \ .set_handle_invalid('keep') - output = vector_assembler.transform(self.input_data_table)[0] - self.assertEqual( - ['id', 'vec', 'num', 'sparse_vec', 'assembled_vec'], - output.get_schema().get_field_names()) + path = os.path.join(self.temp_dir, 'test_save_load_transform_vector_assembler') + vector_assembler.save(path) + vector_assembler = VectorAssembler.load(self.t_env, path) + + output_table = vector_assembler.transform(self.input_data_table)[0] + actual_outputs = [(result[0], result[4]) for result in + self.t_env.to_data_stream(output_table).execute_and_collect()] + + for actual_output in actual_outputs: + if actual_output[0] == 0: + self.assertEqual(self.expected_output_data_1, actual_output[1]) + else: + self.assertEqual(self.expected_output_data_2, actual_output[1])
