[BACKPORT] Implement {DataFrame, Series}.empty (#2155) (#2174)

Author: Xuye (Chris) Qin

mars/_version.py

@@ -15,7 +15,7 @@
 import subprocess
 import os
 
-version_info = (0, 6, 9)
+version_info = (0, 6, 10)
 _num_index = max(idx if isinstance(v, int) else 0
                  for idx, v in enumerate(version_info))
 __version__ = '.'.join(map(str, version_info[:_num_index + 1])) + \

mars/dataframe/base/tests/test_base.py

@@ -902,3 +902,26 @@ def a():
             df.eval("p, q = a + c")
         with self.assertRaises(ValueError):
             df.query("p = a + c")
+
+    def testEmpty(self):
+        # for DataFrame
+        assert from_pandas_df(pd.DataFrame()).empty == pd.DataFrame().empty
+        assert from_pandas_df(pd.DataFrame({})).empty == pd.DataFrame({}).empty
+        assert from_pandas_df(pd.DataFrame({'a':[]})).empty == pd.DataFrame({'a':[]}).empty
+        assert from_pandas_df(pd.DataFrame({'a':[1]})).empty == pd.DataFrame({'a':[1]}).empty
+        assert from_pandas_df(pd.DataFrame({'a':[1], 'b':[2]})).empty == pd.DataFrame({'a':[1], 'b':[2]}).empty
+        assert from_pandas_df(pd.DataFrame(np.empty(shape=(4,0)))).empty == pd.DataFrame(np.empty(shape=(4,0))).empty
+
+        # for Series
+        assert from_pandas_series(pd.Series()).empty == pd.Series().empty
+        assert from_pandas_series(pd.Series({})).empty == pd.Series({}).empty
+        assert from_pandas_series(pd.Series({'a':[]})).empty == pd.Series({'a':[]}).empty
+        assert from_pandas_series(pd.Series({'a':[1]})).empty == pd.Series({'a':[1]}).empty
+
+        # Maybe fail due to lazy evaluation
+        with self.assertRaises(ValueError):
+            a = from_pandas_df(pd.DataFrame(np.random.rand(10, 2)))
+            assert a[a > 0].empty == True
+        with self.assertRaises(ValueError):
+            a = from_pandas_series(pd.Series(np.random.rand(10)))
+            assert a[a > 0].empty == True

mars/dataframe/core.py

@@ -881,6 +881,13 @@ def index(self):
     def axes(self):
         return [self.index]
 
+    @property
+    def empty(self):
+        shape = getattr(self, '_shape')
+        if np.any(np.isnan(shape)):
+            raise ValueError('Tileable object must be executed first')
+        return shape == (0,)
+
     def to_tensor(self, dtype=None):
         from ..tensor.datasource.from_dataframe import from_series
         return from_series(self, dtype=dtype)
@@ -1243,6 +1250,13 @@ def index_value(self):
     def columns_value(self):
         return self._columns_value
 
+    @property
+    def empty(self):
+        shape = getattr(self, '_shape')
+        if np.any(np.isnan(shape)):
+            raise ValueError('Tileable object must be executed first')
+        return (0 in shape)
+
     def to_tensor(self, dtype=None):
         from ..tensor.datasource.from_dataframe import from_dataframe
         return from_dataframe(self, dtype=dtype)

mars/utils.py

@@ -198,27 +198,25 @@ def _get_ports_from_netstat():
     while True:
         p = subprocess.Popen('netstat -a -n -p tcp'.split(), stdout=subprocess.PIPE)
         try:
-            p.wait(5)
-            break
+            outs, _ = p.communicate(timeout=5)
+            outs = to_str(outs).split('\n')
+            occupied = set()
+            for line in outs:
+                if '.' not in line:
+                    continue
+                for part in line.split():
+                    if '.' in part:
+                        _, port_str = part.rsplit('.', 1)
+                        if port_str == '*':
+                            continue
+                        port = int(port_str)
+                        if LOW_PORT_BOUND <= port <= HIGH_PORT_BOUND:
+                            occupied.add(int(port_str))
+                        break
+            return occupied
         except subprocess.TimeoutExpired:
-            p.terminate()
-            continue
-    occupied = set()
-    for line in p.stdout:
-        line = to_str(line)
-        if '.' not in line:
+            p.kill()
             continue
-        for part in line.split():
-            if '.' in part:
-                _, port_str = part.rsplit('.', 1)
-                if port_str == '*':
-                    continue
-                port = int(port_str)
-                if LOW_PORT_BOUND <= port <= HIGH_PORT_BOUND:
-                    occupied.add(int(port_str))
-                break
-    p.stdout.close()
-    return occupied
 
 
 def get_next_port(typ=None, occupy=True):