Spark admite escaneos de columna verdaderos sobre archivos de parquet en S3?

DESCARGO DE RESPONSABILIDAD: No tengo una respuesta definitiva y tampoco quiero actuar como una fuente autorizada, pero he pasado algún tiempo en el soporte de parquet en Spark 2.2+ y espero que mi respuesta pueda ayudarnos a todos a acercarnos a la respuesta correcta.

¿Parquet en S3 evita extraer los datos de las columnas no utilizadas de S3 y solo recupera los trozos de archivo que necesita, o extrae todo el archivo?

Uso Spark 2.3.0-SNAPSHOT que construí hoy desde el maestro .

parquet el formato de fuente de datos es manejado por ParquetFileFormat que es un FileFormat .

Si estoy en lo correcto, la parte de lectura es manejada por el método buildReaderWithPartitionValues (que anula los FileFormat's).

buildReaderWithPartitionValues se utiliza exclusivamente cuando se solicita el operador físico FileSourceScanExec para los llamados RDD de entrada que son en realidad un solo RDD para generar filas internas cuando WholeStageCodegenExec es ejecutar.

Dicho esto, creo que revisar lo que hace buildReaderWithPartitionValues puede acercarnos a la respuesta final.

Cuando nos fijamos en la línea usted puede estar seguro de que estamos en el camino correcto.

// Intente empujar hacia abajo los filtros cuando filtro push-down está habilitado.

Esa ruta de código depende de spark.sql.parquet.filterPushdown la propiedad Spark que está activada por defecto.

Spark.SQL.parquet.filterPushdown Permite Parquet filtrar la optimización push-down cuando se establece en true.

Que nos lleva a parquet-hadoop ParquetInputFormat.setFilterPredicate iff los filtros están definidos.

if (pushed.isDefined) {
  ParquetInputFormat.setFilterPredicate(hadoopAttemptContext.getConfiguration, pushed.get)
}

El código se vuelve más interesante un poco más tarde cuando se utilizan los filtros cuando el código vuelve a parquet-mr (en lugar de usar el llamado lector de decodificación de parquet vectorizado). Esa es la parte que realmente no entiendo (excepto lo que puedo ver en el código).

Tenga en cuenta que el lector de decodificación de parquet vectorizado está controlado por la propiedad Spark spark.sql.parquet.enableVectorizedReader que está activada de forma predeterminada.

CONSEJO: Para saber qué parte de la expresión if se usa, habilite el nivel de registro DEBUG para org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat logger.

Para ver todos los filtros empujados hacia abajo, puede activar INFO el nivel de registro de org.apache.spark.sql.execution.FileSourceScanExec logger. Deberías ver lo siguiente en los registros :

INFO Pushed Filters: [pushedDownFilters]

Espero que si no está cerca de ser una respuesta definitiva que ha ayudado a un little y alguien lo recoge donde lo dejé para hacerlo pronto. La esperanza muere el último :)

No, la pushdown de predicados no es totalmente compatible. Esto, por supuesto, depende de:

• Caso de uso específico
• Versión Spark
• Tipo y versión del conector S3

Para verificar su caso de uso específico, puede habilitar el nivel de registro de depuración en Spark y ejecutar su consulta. Luego, puede ver si hay" seeks " durante las solicitudes S3 (HTTP), así como cuántas solicitudes se enviaron realmente. Algo como esto:

17/06/13 05:46:50 DEBUG wire: http-outgoing-1 >> "GET /test/part-00000-b8a8a1b7-0581-401f-b520-27fa9600f35e.snappy.parquet HTTP/1.1[\r][\n]" .... 17/06/13 05:46:50 DEBUG wire: http-outgoing-1 << "Content-Range: bytes 0-7472093/7472094[\r][\n]" .... 17/06/13 05:46:50 DEBUG wire: http-outgoing-1 << "Content-Length: 7472094[\r][\n]"

Aquí está el ejemplo de una informe de problemas que se abrió recientemente debido a la incapacidad de Spark 2.1 para calcular COUNT(*) de todas las filas de un conjunto de datos basado en metadatos almacenados en el archivo Parquet: https://issues.apache.org/jira/browse/SPARK-21074

Parquet reader de spark es como cualquier otro formato de entrada,

1. Ninguno de los InputFormat tiene nada especial para S3. Los formatos de entrada pueden leer desde LocalFileSystem, Hdfs y S3 sin ninguna optimización especial hecha para eso.

2. Parquet InputFormat dependiendo de las columnas que pregunte leerá selectivamente las columnas por usted .

3. Si desea estar completamente seguro (aunque push down predicates funciona en la última versión de spark) seleccione manualmente columnas y escribir la transformación y las acciones, en lugar de depender de SQL