¿Cómo guardar un marco de datos de pandas enorme en hdfs?

Bienvenido a nuestra página web, en este sitio vas a hallar la respuesta que buscabas.

Solución:

Lo que significa tener un marco de datos de pandas que transformo en chispa con la ayuda de pyarrow.

pyarrow.Table.fromPandas es la función que buscas:

Table.from_pandas(type cls, df, bool timestamps_to_ms=False, Schema schema=None, bool preserve_index=True)

Convert pandas.DataFrame to an Arrow Table

import pyarrow as pa

pdf = ...  # type: pandas.core.frame.DataFrame
adf = pa.Table.from_pandas(pdf)  # type: pyarrow.lib.Table

El resultado se puede escribir directamente en Parquet/HDFS sin pasar datos a través de Spark:

import pyarrow.parquet as pq

fs  = pa.hdfs.connect()

with fs.open(path, "wb") as fw
    pq.write_table(adf, fw)

Ver también

• Respuesta de @WesMcKinney para leer archivos de parquet de HDFS usando PyArrow.
• Leer y escribir el formato Apache Parquet en el pyarrow documentación.
• Conectividad nativa del sistema de archivos Hadoop (HDFS) en Python

Notas de chispa:

Además, dado que Spark 2.3 (maestro actual), Arrow es compatible directamente en createDataFrame (SPARK-20791: use la flecha de Apache para mejorar Spark createDataFrame desde Pandas.DataFrame). Usa SparkContext.defaultParallelism para calcular el número de trozos para que pueda controlar fácilmente el tamaño de los lotes individuales.

Por fin defaultParallelism se puede usar para controlar el número de particiones generadas usando estándar _convert_from_pandas, reduciendo efectivamente el tamaño de las porciones a algo más manejable.

Desafortunadamente, es poco probable que estos resuelvan sus problemas de memoria actuales. Ambos dependen de parallelize, por lo tanto, almacene todos los datos en la memoria del nodo del controlador. Cambiar a Arrow o ajustar la configuración solo puede acelerar el proceso o abordar las limitaciones del tamaño del bloque.

En la práctica, no veo ninguna razón para cambiar a Spark aquí, siempre que use Pandas locales DataFrame como entrada. El cuello de botella más grave en este escenario es la E/S de la red del controlador y la distribución de datos no solucionará eso.

De https://issues.apache.org/jira/browse/SPARK-6235

Compatibilidad con la paralelización de marcos de datos R de más de 2 GB

esta resuelto.

De https://pandas.pydata.org/pandas-docs/stable/r_interface.html

Convertir tramas de datos en objetos R

puede convertir un marco de datos de pandas en un marco de datos R

Entonces, ¿quizás la transformación pandas -> R -> Spark -> hdfs?

Otra forma es convertir su marco de datos de pandas en un marco de datos de chispa (usando pyspark) y guardarlo en hdfs con el comando guardar. ejemplo

    df = pd.read_csv("data/as/foo.csv")
    df[['Col1', 'Col2']] = df[['Col2', 'Col2']].astype(str)
    sc = SparkContext(conf=conf)
    sqlCtx = SQLContext(sc)
    sdf = sqlCtx.createDataFrame(df)

Aquí astype cambia el tipo de su columna de object a string. Esto lo salva de una excepción planteada de otro modo, ya que Spark no pudo descifrar el tipo de pandas object. Pero asegúrese de que estas columnas realmente sean del tipo string.

Ahora para guardar su df en hdfs:

    sdf.write.csv('mycsv.csv')

Si te ha sido de ayuda nuestro post, nos gustaría que lo compartas con más programadores de esta forma contrubuyes a extender esta información.

---

---

Utiliza Nuestro Buscador