Koalas: Fails to read from parquet table with flat schema and fields containing "."

[sadikovi]

I observed that when loading parquet data from a table that has a mix of normal fields and fields that have . in their name:

Sample data:

from pyspark.sql.functions import lit
spark.range(1).withColumn("a.b", lit(1)).withColumn("a_b", lit(2)).coalesce(1).write.parquet("/mnt/ivan/sample.parquet")

Code works when accessing "a_b":

df = ks.read_parquet("/mnt/ivan/sample.parquet")
df['a_b']

# Out[9]: 
# 0    2
# Name: a_b, dtype: int32

Accessing "a.b" column fails:

df = ks.read_parquet("/mnt/ivan/sample.parquet")
df['a.b']

# KeyError: 'a.b'
# ---------------------------------------------------------------------------
# KeyError                                  Traceback (most recent call last)
# <command-3188040896828876> in <module>()
#       1 df = ks.read_parquet("/mnt/ivan/sample.parquet")
# ----> 2 df['a.b']
# 
# /databricks/python/lib/python3.5/site-packages/databricks/koalas/frame.py in __getitem__(self, key)
#     448 
#     449     def __getitem__(self, key):
# --> 450         return self._pd_getitem(key)
#     451 
#     452     def __setitem__(self, key, value):
# 
# /databricks/python/lib/python3.5/site-packages/databricks/koalas/frame.py in _pd_getitem(self, key)
#     427                 return Series(self._sdf.__getitem__(key), self, self._metadata.index_info)
#     428             except AnalysisException:
# --> 429                 raise KeyError(key)
#     430         if np.isscalar(key) or isinstance(key, (tuple, string_types)):
#     431             raise NotImplementedError(key)
# 
# KeyError: 'a.b'

Escaping column name with "`" also does not work:

df = ks.read_parquet("/mnt/ivan/sample.parquet")
df['`a.b`']

# Out[11]: <repr(<databricks.koalas.series.Series at 0x7fe1347c53c8>) failed: KeyError: 'a.b'>

[sadikovi]

I will work on it, try fixing it as Tim suggested.

[rxin]

I just ran into the same issue. Please fix it @sadikovi :)

[rxin]

BTW the one I ran into was with CSV data. One of the column names have dot in it. And calling dtypes would return a bad exception, whereas calling to_spark().dtypes would be fine.

---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
/anaconda3/envs/koalas-dev-env/lib/python3.6/site-packages/pyspark/sql/utils.py in deco(*a, **kw)
     62         try:
---> 63             return f(*a, **kw)
     64         except py4j.protocol.Py4JJavaError as e:

/anaconda3/envs/koalas-dev-env/lib/python3.6/site-packages/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
    327                     "An error occurred while calling {0}{1}{2}.\n".
--> 328                     format(target_id, ".", name), value)
    329             else:

Py4JJavaError: An error occurred while calling o610.apply.
: org.apache.spark.sql.AnalysisException: syntax error in attribute name: Check here if you want to join the Databricks mailing list to get news and updates about Apache Spark and our products.;
    at org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute$.e$1(unresolved.scala:152)
    at org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute$.parseAttributeName(unresolved.scala:171)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.resolveQuoted(LogicalPlan.scala:121)
    at org.apache.spark.sql.Dataset.resolve(Dataset.scala:221)
    at org.apache.spark.sql.Dataset.col(Dataset.scala:1268)
    at org.apache.spark.sql.Dataset.apply(Dataset.scala:1235)
    at sun.reflect.GeneratedMethodAccessor16.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:282)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748)


During handling of the above exception, another exception occurred:

AnalysisException                         Traceback (most recent call last)
~/workspace/koalas/databricks/koalas/frame.py in _pd_getitem(self, key)
   1475             try:
-> 1476                 return Series(self._sdf.__getitem__(key), self, self._metadata.index_info)
   1477             except AnalysisException:

/anaconda3/envs/koalas-dev-env/lib/python3.6/site-packages/pyspark/sql/dataframe.py in __getitem__(self, item)
   1278         if isinstance(item, basestring):
-> 1279             jc = self._jdf.apply(item)
   1280             return Column(jc)

/anaconda3/envs/koalas-dev-env/lib/python3.6/site-packages/py4j/java_gateway.py in __call__(self, *args)
   1256         return_value = get_return_value(
-> 1257             answer, self.gateway_client, self.target_id, self.name)
   1258 

/anaconda3/envs/koalas-dev-env/lib/python3.6/site-packages/pyspark/sql/utils.py in deco(*a, **kw)
     68             if s.startswith('org.apache.spark.sql.AnalysisException: '):
---> 69                 raise AnalysisException(s.split(': ', 1)[1], stackTrace)
     70             if s.startswith('org.apache.spark.sql.catalyst.analysis'):

AnalysisException: 'syntax error in attribute name: Check here if you want to join the Databricks mailing list to get news and updates about Apache Spark and our products.;'

During handling of the above exception, another exception occurred:

KeyError                                  Traceback (most recent call last)
<ipython-input-7-5cc0934cc03c> in <module>
----> 1 df.dtypes

~/workspace/koalas/databricks/koalas/frame.py in dtypes(self)
   1113         dtype: object
   1114         """
-> 1115         return pd.Series([self[col].dtype for col in self._metadata.column_fields],
   1116                          index=self._metadata.column_fields)
   1117 

~/workspace/koalas/databricks/koalas/frame.py in <listcomp>(.0)
   1113         dtype: object
   1114         """
-> 1115         return pd.Series([self[col].dtype for col in self._metadata.column_fields],
   1116                          index=self._metadata.column_fields)
   1117 

~/workspace/koalas/databricks/koalas/frame.py in __getitem__(self, key)
   1500 
   1501     def __getitem__(self, key):
-> 1502         return self._pd_getitem(key)
   1503 
   1504     def __setitem__(self, key, value):

~/workspace/koalas/databricks/koalas/frame.py in _pd_getitem(self, key)
   1476                 return Series(self._sdf.__getitem__(key), self, self._metadata.index_info)
   1477             except AnalysisException:
-> 1478                 raise KeyError(key)
   1479         if np.isscalar(key) or isinstance(key, (tuple, string_types)):
   1480             raise NotImplementedError(key)

KeyError: 'Check here if you want to join the Databricks mailing list to get news and updates about Apache Spark and our products.'

[sadikovi]

Thanks for letting me know. Will do!

[rxin]

@sadikovi any eta?

[sadikovi]

I am still working on the problem, found there are quite a few examples for which I would need to check my patch. Will submit PR this week, apologies for the delay.

[rxin]

@sadikovi any update?

[sadikovi]

I have been busy with some other stuff and have not spent much time on fixing this issue, would appreciate if someone could pick it up.

We should come up with a way that handles the following cases:

• top level primitive column a.b
• complex field a.b
• two fields: primitive a.b and complex a.b

I tried having a simple function to surround column names with backticks:

def sdf_getitem(sdf, key):
    cols = sdf.columns
    if "." in key and key in cols:
        return spark.functions.col("`%s`" % (key,))
    return key

And use it as following:

-                return Series(self._sdf.__getitem__(key), anchor=self,
+                return Series(sdf_getitem(self._sdf, key), anchor=self,

Unfortunately, this does not quite work when DataFrames have complex field names or field names defined as references of DataFrame, e.g. df[a.b.], or when using select:

sdf = self._sdf.select(*exprs)

Another idea I explored was updating the code to keep columns instead of column names, but found it to be potentially quite complicated and abandoned it.

Let me know if there are other ways of solving this I have not considered.

[icexelloss]

We have the same issue with Spark, i.e., internally we have users that historically names their columns with the parameter, e.g., "result_arg1_0.9_arg2_0.8" etc. We have to patch Spark internally to allow that.

I wonder if this is something that should be handled at the Spark layer?

[sadikovi]

Spark can handle top level columns with "." in them by escaping names with backticks/backquotes; however, this does not work in koalas, since the code passes column names without escaping them properly.

[rxin]

cc @cloud-fan @HyukjinKwon can you two come up with a proposal to fix this? Seems pretty complicated and would require thinking through nested type support.

[cloud-fan]

I'm not very familiar with koalas, but this case works fine in Spark

scala> df.printSchema
root
 |-- id: long (nullable = true)
 |-- a.b: integer (nullable = true)
 |-- a: struct (nullable = true)
 |    |-- b: integer (nullable = true)


scala> df("a.b")
res5: org.apache.spark.sql.Column = a.b AS `b`

scala> df("`a.b`")
res6: org.apache.spark.sql.Column = a.b

[HyukjinKwon]

Let me take this

[HyukjinKwon]

Here's what's going on now:

scala> val df = spark.range(1).toDF("a.b")
df: org.apache.spark.sql.DataFrame = [a.b: bigint]

scala> df("a.b")
org.apache.spark.sql.AnalysisException: Cannot resolve column name "a.b" among (a.b);
  at org.apache.spark.sql.Dataset.$anonfun$resolve$1(Dataset.scala:233)
  at scala.Option.getOrElse(Option.scala:138)
  at org.apache.spark.sql.Dataset.resolve(Dataset.scala:233)
  at org.apache.spark.sql.Dataset.col(Dataset.scala:1314)
  at org.apache.spark.sql.Dataset.apply(Dataset.scala:1281)
  ... 47 elided

a.b is being resolved as nested access