Как вывести сгенерированный Java-код на стандартный вывод?

Question

Используя DataFrames в Apache Spark 2. +, есть ли способ получить базовые RDD и вывести сгенерированный Java-код на консоль?

Answer 1

Это можно сделать с помощью QueryExecution.debug.codegen. Это значение доступно в Dataframe / Dataset через .queryExecution (это «API разработчика», то есть нестабильный, подверженный поломке и поэтому должен использоваться только для отладки). Это работает на Spark 2.4.0, и из кода кажется, что оно должно работать с 2.0.0 (или более):

scala> val df = spark.range(1000)
df: org.apache.spark.sql.Dataset[Long] = [id: bigint]

scala> df.queryExecution.debug.codegen
Found 1 WholeStageCodegen subtrees.
== Subtree 1 / 1 ==
*(1) Range (0, 1000, step=1, splits=12)

Generated code:
/* 001 */ public Object generate(Object[] references) {
/* 002 */   return new GeneratedIteratorForCodegenStage1(references);
/* 003 */ }
/* 004 */
/* 005 */ // codegenStageId=1
/* 006 */ final class GeneratedIteratorForCodegenStage1 extends org.apache.spark.sql.execution.BufferedRowIterator {
/* 007 */   private Object[] references;
/* 008 */   private scala.collection.Iterator[] inputs;
/* 009 */   private boolean range_initRange_0;
/* 010 */   private long range_number_0;
/* 011 */   private TaskContext range_taskContext_0;
/* 012 */   private InputMetrics range_inputMetrics_0;
/* 013 */   private long range_batchEnd_0;
/* 014 */   private long range_numElementsTodo_0;
/* 015 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[] range_mutableStateArray_0 = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[1];

...

/* 104 */       ((org.apache.spark.sql.execution.metric.SQLMetric) references[0] /* numOutputRows */).add(range_nextBatchTodo_0);
/* 105 */       range_inputMetrics_0.incRecordsRead(range_nextBatchTodo_0);
/* 106 */
/* 107 */       range_batchEnd_0 += range_nextBatchTodo_0 * 1L;
/* 108 */     }
/* 109 */   }
/* 110 */
/* 111 */ }

Answer 2

Вот способ вывода сгенерированного кода, возможно, есть другие способы сделать это:

import org.apache.spark.sql.execution.command.ExplainCommand

val explain = ExplainCommand(df.queryExecution.logical, codegen=true)
spark.sessionState.executePlan(explain).executedPlan.executeCollect().foreach {
  r => println(r.getString(0))
}