https://github.com/alibaba/Alink/blob/master/docs/pyalink/pyalink-dataframe.md

與 Dataframe 互操作

PyAlink 提供了與pandas DataFrame的互轉操作，能夠方便地使用 Python 生態中已有的強大工具。 pandas 的 DataFrame 是 Python 已有生態中表示、儲存二維表的十分理想的選擇，DataFrame 自身提供了一定的資料處理與視覺化能力，同時又可以方便的轉換為 Python 中的其他資料結構。

BatchOperator 轉 Dataframe

對於各個BatchOperator，提供了collectToDataframe()

source = CsvSourceBatchOp()\
    .setSchemaStr("sepal_length double, sepal_width double, petal_length double, petal_width double, category string")\
    .setFilePath("https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv")
res = source.select(["sepal_length", "sepal_width"])
df = res.collectToDataframe()
# Operations with df
res.print()
 

在BatchOperator中也提供了靜態方法collectoToDataframes可以對多個BatchOperator同時進行轉換。 每一次單個BatchOperator呼叫collectoToDataframe都會導致作業圖進行一次計算，多次呼叫時將有效率問題，更重要的是，有些演算法在多次呼叫時可能結果不一致，例如取樣操作等。 此時可以使用這個靜態方法來解決這些問題。

source = CsvSourceBatchOp() \
    .setSchemaStr(
    "sepal_length double, sepal_width double, petal_length double, petal_width double, category string") \
    .setFilePath("https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv")

split = SplitBatchOp().setFraction(0.1).linkFrom(source)
[d0, d1, d2] = collectToDataframes(source, split, split.getSideOutput(0))
print(d0, d1, d2)
 

Dataframe 轉 BatchOperator/StreamOperator

對於使用者擁有的 Dataframe 資料，PyAlink 也提供了向 BatchOperator/StreamOperator 轉換的方法。 具體說來，BatchOperator/StreamOperator中都具有靜態方法fromDataframe(df, schemaStr)，其中df是 DataFrame 資料，schemaStr是由資料列名和型別構成的 schema 字串。

schema = "f_string string,f_long long,f_int int,f_double double,f_boolean boolean"
op = BatchOperator.fromDataframe(df, schema)
op.print()

op = StreamOperator.fromDataframe(df, schema)
op.print(key="op")
StreamOperator.execute()

同時，PyAlink 也提供了靜態方法來進行轉換：dataframeToOperator(df, schemaStr, opType)，這裡df和schemaStr引數與上文相同，opType取值為"batch"或"stream"。

使用注意

• 從 BatchOperator 向 Dataframe 轉換時：如果 BatchOperator 所儲存的資料中有空值null，那麼轉換後的 Dataframe 將無法保證完全轉換為 BatchOperator 的列型別。此時，對應的列需要手動進行處理。
• 從 Dataframe 向 BatchOperator 轉換時：如果 Dataframe 中有空值時，需要預先替換為None，這樣才能正確地進行之後的處理。
• 當資料量大時，由於資料轉換需要進行大量的 Python 與 Java 資料之間的交換，效率將會受到影響。