fix bug of SparkSql Operator log going to infinite loop. (#19449)

apache · Nov 24, 2021 · 79b30ff · 79b30ff
1 parent ac752e7
commit 79b30ff
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 1 deletion.
diff --git a/airflow/providers/apache/spark/hooks/spark_sql.py b/airflow/providers/apache/spark/hooks/spark_sql.py
@@ -182,7 +182,9 @@ def run_query(self, cmd: str = "", **kwargs: Any) -> None:
         """
         spark_sql_cmd = self._prepare_command(cmd)
 
-        self._sp = subprocess.Popen(spark_sql_cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kwargs)
+        self._sp = subprocess.Popen(
+            spark_sql_cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True, **kwargs
+        )
 
         for line in iter(self._sp.stdout):  # type: ignore
             self.log.info(line)

diff --git a/tests/providers/apache/spark/hooks/test_spark_sql.py b/tests/providers/apache/spark/hooks/test_spark_sql.py
@@ -127,6 +127,7 @@ def test_spark_process_runcmd(self, mock_popen):
             ],
             stderr=-2,
             stdout=-1,
+            universal_newlines=True,
         )
 
     @patch('airflow.providers.apache.spark.hooks.spark_sql.subprocess.Popen')
@@ -156,6 +157,7 @@ def test_spark_process_runcmd_with_str(self, mock_popen):
             ],
             stderr=-2,
             stdout=-1,
+            universal_newlines=True,
         )
 
     @patch('airflow.providers.apache.spark.hooks.spark_sql.subprocess.Popen')
@@ -185,6 +187,7 @@ def test_spark_process_runcmd_with_list(self, mock_popen):
             ],
             stderr=-2,
             stdout=-1,
+            universal_newlines=True,
         )
 
     @patch('airflow.providers.apache.spark.hooks.spark_sql.subprocess.Popen')