Memsql::Streamliner Python 转换答案

【问题标题】：Memsql::Streamliner Python TransformMemsql::Streamliner Python 转换
【发布时间】：2016-04-13 18:19:42
【问题描述】：

我正在使用 Memsql::Streamliner::Transform (Python) 实用程序。必须重写一个转换方法才能提供自定义转换功能。

def transform(self, sql_context, dataframe, logger):

dataframe.column[0] 是一个字节数组（JSON 字符串）。

如何将字节数组转换为具有命名列的 DataFrame？

目标：访问转换后的 DataFrame 中的各个列。

【问题讨论】：

标签： python pyspark spark-dataframe singlestore

【解决方案1】：

您可以使用dataframe.rdd 访问底层rdd 并对其进行映射以将每个字节字符串转换为包含您的列的列表。您可以通过将列列表作为第二个参数提供给 createDataframe，将生成的 rdd 转换回具有命名列的数据框。

类似下面的东西应该可以工作：

def parse(row):
   bytestring = row[0]
   json_data = convert_bytes_and_parse_json(bytestring)
   return [ json_data["mycolumn1"], json_data["mycolumn2"] ]

parsedRDD = dataframe.rdd.map(parse)
parsedDf = sql_context.createDataframe(parsedRDD, ["mycolumn1", "mycolumn2"])

# now you can access columns by name
parsedDf.select(parsedDf["mycolumn1"])

【讨论】：

您的示例有所帮助。谢谢。但是，这些列似乎是按顺序浮动的。如果我每次运行代码时访问 parsedDf[0]、parsedDF[1] 等，不同的列会出现在 [0] 点等。想法？
您是在 parse 函数中返回列表还是返回字典？我没有对此进行测试，但是没有保留字典中的键顺序，因此它可能会根据您的 rdd 中的第一个字典来选择列的顺序。在任何情况下，您都可以使用 df["colname"] 直接访问您的列，它在后台运行 df.select("colname")。