在 Python 中通过 pyarrow 和（或）pandas 来构造镶木地板的列

Question

我希望你们中的一些人能抽出一点时间来帮助像我这样的初学者。我整整一周都在处理这项任务，但找不到解决方案。我理解并且完全没问题，我必须学习我使用的每个包及其组合才能找到正确的解决方案。

完整的任务是将 5 列（1000 行）组合成 1 个结构列，并在 parquet 中将其存储/转换为一行（1000 列）。但我坚持将 5 列合并为 1 个结构列的问题。

最初，我收到以下列：columns=['date', 'bidopen', 'bidclose', 'bidhigh', 'bidlow', 'tickqty']。我不需要“日期”作为结构的一部分。

我尝试了什么：

import pyarrow as pa
import pyarrow.parquet as pq
import pandas as pd

选项 1 - 带熊猫的字典

df = pd.read_csv('original.csv')
df2 = df.drop(columns=['date'])
df3 = df2.to_dict('records')

我无法通过 pandas 将 dict 保存到 csv 或 parquet - 以下 2 个命令导致向后转换到 pandas 数据帧并分别保存列。

pd.DataFrame(df3).to_csv('test_parquet.csv', index=False)
pd.DataFrame(df3).to_parquet('test2.parquet')

如果我可以将字典用作数据框，那么接下来我将使用 pandas.DataFrame.pivot 将行转换为列。接下来，我尝试将 dict 转换为 pyarrow 表（似乎我还可以将条目保存在列（1 行）中）。

table = pa.Table.from_pydict({'data', pa.array(df3)})

在上面的行之后我有一个错误，我找不到解决方案（TypeError: unhashable type: 'pyarrow.lib.StructArray'）。下一步是通过 pyarrow 将表格保存到 parquet 中。

选项 2 - 由 pyarrow 构造

在这里我尝试在 parquet 内部工作以更改架构（或写入新架构）

df = pd.read_csv('original.csv')
df = df.drop(columns=['date'])
df.to_parquet('test.parquet')
table = pq.read_table('test.parquet', columns=['bidopen', 'bidclose', 'bidhigh', 'bidlow', 'tickqty'])

在这里，我阅读了 parquet 的架构以查看每一列的 DataType。下面我设置了新的架构：

struct = pa.struct([
    pa.field('bidopen', pa.float64()),
    pa.field('bidclose', pa.float64()),
    pa.field('bidhigh', pa.float64()),
    pa.field('bidlow', pa.float64()),
    pa.field('tickqty', pa.int64())
])
fields = ([pa.field('data', pa.list_(struct))])
schema = pa.schema(fields)
writer = pq.ParquetWriter('test2.parquet', schema)
writer.write_table(table)
writer.close()

我遇到了一个错误，我也找不到解决方案（ValueError：表架构与用于创建文件的架构不匹配：...），因为我认为它会保存到新提供的架构中。

选项 3 - pyarrow cast

#(the upper part is from the Option 2)
...
schema = pa.schema(fields)
table2 = table.cast(schema)
writer = pq.ParquetWriter('test2.parquet', schema)
writer.write_table(table2)
writer.close()

我收到另一个错误（ValueError：目标架构的字段名称与表的字段名称不匹配：）。在这里我说 - 来吧，我正在做演员，因为模式不一样......这没有帮助。

选项 4 - 从 pandas 加载到 pyarrow 以稍后将其保存到 parquet 时尝试更改架构

arrays = [['data','data','data','data','data'],['bidopen', 'bidclose','bidhigh','bidlow','tickqty']]
tuples = list(zip(*arrays))
index = pd.MultiIndex.from_tuples(tuples)
df2 = pd.DataFrame(df.values[:, 1:], columns=index)
pa.Schema.from_pandas(df2)

这里我遇到了一个错误（AttributeError: 'list' object has no attribute 'columns'），我也找不到解决方案。

选项 5 - pyspark

这对我来说是最大的麻烦，因为我花了大约 3 天的时间来“学习”它，因为它应该能够转换为结构和旋转。但是，我后来发现，如果没有额外的软件包，我无法在我的 Win10 上通过 pyspark 将数据保存到 parquet 中：Hadoop 和 Java SDK（它不是免费使用的）。因此，我停止了进一步的开发。

Answer 1

对于您问题的第一部分，您可以这样做（注意， StructArray.from_arrays 需要数组，因此您需要展平分块数组）：

fields, arrs = [], []
for column_index in range(table.num_columns):
    fields.append(table.field(column_index))
    arrs.append(table.column(column_index).flatten()[0].chunks[0])
struct_array = pa.StructArray.from_arrays(arrs, fields=fields)
print(struct_array)
print(struct_array.to_pylist())

样本输出：

-- is_valid: all not null
-- child 0 type: double
  [
    1.1,
    2.2
  ]
-- child 1 type: double
  [
    3.3,
    4.4
  ]
-- child 2 type: double
  [
    5.5,
    6.6
  ]
-- child 3 type: double
  [
    7.7,
    8.8
  ]
-- child 4 type: int64
  [
    9,
    10
  ]
[{'bidopen': 1.1, 'bidclose': 3.3, 'bidhigh': 5.5, 'bidlow': 7.7, 'tickqty': 9}, {'bidopen': 2.2, 'bidclose': 4.4, 'bidhigh': 6.6, 'bidlow': 8.8, 'tickqty': 10}]

如果这是您对问题第二部分的要求，我认为 pyarrow 不能转置。您可以使用 pandas 进行转置，但它会是另一个副本。

df = pa.Table.from_arrays([struct_array], ['data']).to_pandas()
print(df.transpose())

样本输出：

                                                      0  \
data  {'bidopen': 1.1, 'bidclose': 3.3, 'bidhigh': 5...   

                                                      1  
data  {'bidopen': 2.2, 'bidclose': 4.4, 'bidhigh': 6...  

​

在这种情况下，输出总是一个包含 N 列的单行表，每个单元格都是一个结构。