だいぶ、小ネタなのですが…
PySparkを使ってデータ操作していたとき、カラム名に「:」が入ってて困った話を今回はしてみます。

サンプルデータフレームを用意

今日の晩御飯はすき焼きだったので、牛肉にちなんで、牛肉の産地表と、ブランド肉の名称表という2つのデータフレームを用意し、それを結合する、というものをやってみます。

meat_from = [
    {"hoge:id": "A1", "from": "kobe"}, 
    {"hoge:id": "A2", "from": "matsuszaka"}
    ]
meat_name= [
     {"foo:id": "A1", "name": "神戸牛"}, 
     {"foo:id": "A2", "name": "松坂牛"}
    ]   
meat_from_sdf = spark.createDataFrame(meat_from)
meat_name_sdf = spark.createDataFrame(meat_name)

んで、

meat_from_sdf.show()
>>>
+----------+-------+
|      from|hoge:id|
+----------+-------+
|      kobe|     A1|
|matsuszaka|     A2|
+----------+-------+

meat_name_sdf.show()
>>>
+------+------+
|foo:id|  name|
+------+------+
|    A1|神戸牛 |
|    A2|松坂牛 |
+------+------+

(列の順番が雑なのは、ごめんなさい)

よく見ると、両表にidらしきカラムがありますが、それぞれ「hoge:id」「foo:id」と、コロンが付いています。
名前は違いますが、同じ意味のデータを持つと仮定して、このカラムたちをキーに、のちほど結合処理をしてみます。

単純なselectは行ける

カラム名にコロンが付いていることはあまり見慣れないですが、試しにカラムをselectしてみる。

selected_meat_from_sdf = meat_from_sdf.select("hoge:id")
selected_meat_from_sdf.show()
>>>
+-------+
|hoge:id|
+-------+
|     A1|
|     A2|
+-------+

ふーん、なんの問題もなくselectできるじゃん。
じゃあさっさとjoinしちゃお！と思っていた自分は、甘かった…

joinの方法はいろいろとある

Pyspark DataFrame同士のjoin処理はいろいろとあります。
SQLチックな動きが得意な印象がありますが、まさにそのとおりで、DataFrameのままjoinするのもよし、View化して全てSQLで書くもよし。

このまとめ記事をよく参考にしています。

今回は、View化するのも面倒なので、DataFrameのままjoin処理をやってみます。

joined_sdf = (
    meat_from_sdf
    .join(
        meat_name_sdf,
        meat_from_sdf.hoge:id ==  meat_name_sdf.foo:id,
        how = "inner"
    )
)
joined_sdf.show()
>>>
  File "<ipython-input-13-1f34e2fef92a>", line 5
    meat_from_sdf.hoge:id ==  meat_name_sdf.foo:id,
                      ^
SyntaxError: invalid syntax

インタラクティブに怒られてしまいました。
どうやら、単体でselectするのはイケるが、結合条件を書くところで、コロンが付いていると不具合があるようです。

<sdf_1>.<join_based_column_name_1> == <sdf_2>.<join_based_column_name_2>

手っ取り早いのは、改名

どうするか色々考えたものの、結局、手っ取り早いのは該当カラムを改名することでした。

Pysparkの面白いのが、その改名方法もイロイロあること。
身近に働いている方が、selectExprをよく使っているので、試してみる。

renamed_meat_from_sdf =  meat_from_sdf.selectExpr("hoge:id as hoge_id", "from") 

>>>
ParseException                            Traceback (most recent call last)
<ipython-input-15-baf465d759fb> in <module>
----> 1 renamed_meat_from_sdf =  meat_from_sdf.selectExpr("hoge:id as hoge_id", "from")
      2 renamed_meat_name_sdf =  meat_name_sdf.selectExpr("foo:id as foo_id", "from")

2 frames
/usr/local/lib/python3.7/dist-packages/pyspark/sql/utils.py in deco(*a, **kw)
    194                 # Hide where the exception came from that shows a non-Pythonic
    195                 # JVM exception message.
--> 196                 raise converted from None
    197             else:
    198                 raise

ParseException: 
Syntax error at or near ':'(line 1, pos 4)

== SQL ==
hoge:id as hoge_id
----^^^

これも怒られてしまいました。

おとなしく、select内で.aliasを使って改名することにします。

renamed_meat_from_sdf =  (
    meat_from_sdf
    .select(
        col("hoge:id").alias("hoge_id"),
        col("from")
    )
)
renamed_meat_from_sdf.show()
>>>
+-------+----------+
|hoge_id|      from|
+-------+----------+
|     A1|      kobe|
|     A2|matsuszaka|
+-------+----------+

今回は出来ました。

同じく、もうひとつのDataFrameにも処理して、join処理。ついでにselectして、id系カラムはひとつに絞る。

joined_sdf = (
    renamed_meat_from_sdf
    .join(
        renamed_meat_name_sdf,
        renamed_meat_from_sdf.hoge_id ==  renamed_meat_name_sdf.foo_id,
        how = "inner"
    )
    .select(
        col("hoge_id").alias("id"),
        col("from"),
        col("name")
    )
)
joined_sdf.show()
>>>
+---+----------+------+
| id|      from|  name|
+---+----------+------+
| A1|      kobe|神戸牛 |
| A2|matsuszaka|松坂牛 |
+---+----------+------+

成功しました。

View化してSQLで結合しようとしても…

結果は同じでした。

meat_from_sdf.createOrReplaceTempView("meat_from_tmpview")
meat_name_sdf.createOrReplaceTempView("meat_name_tmpview")

joined_sdf = spark.sql("""
select * from meat_from_tmpview 
inner join meat_name_tmpview 
on meat_from_tmpview.hoge:id = meat_name_tmpview.foo:id
""")
joined_sdf.show()

>>>
ParseException                            Traceback (most recent call last)
<ipython-input-28-35e4897329db> in <module>
----> 1 a = spark.sql("""select * from meat_from_tmpview inner join meat_name_tmpview on meat_from_tmpview.hoge:id =  meat_name_tmpview.foo:id""")
      2 a.show()

2 frames
/usr/local/lib/python3.7/dist-packages/pyspark/sql/utils.py in deco(*a, **kw)
    194                 # Hide where the exception came from that shows a non-Pythonic
    195                 # JVM exception message.
--> 196                 raise converted from None
    197             else:
    198                 raise

ParseException: 
Syntax error at or near ':'(line 1, pos 86)

== SQL ==
select * from meat_from_tmpview inner join meat_name_tmpview on meat_from_tmpview.hoge:id =  meat_name_tmpview.foo:id
--------------------------------------------------------------------------------------^^^

うーん、やっぱり、コロンがあるとだめみたいです。

まとめ

カラムにコロンが入っていたら、それをraw dataとして保持するぶんには良いですが、データマートを作る段階でコネコネ・ジョイン等する場合は、おとなしく改名するほうが良さそうです。