when we call persist on dataframe and specify storage level to be MEMORY_ONLY or MEMORY_ONLY_SER what is actually serialized/de-serialized?

Question

I was reading about tungsten engine in Spark and figured out when we use dataframe Spark internally create a compact binary format that represent data and apply transformation chain on that compact binary format. in the Spark in Action book MEMORY_ONLY and MEMORY_ONLY_SER are defined like this:

MEMORY_ONLY:‌ This is the default level. It will store the RDD composing the dataframe as deserialized Java objects in the JVM. If the RDD does not fit in memory, Spark will not cache the partitions; Spark will recompute as needed. You will not be notified.
MEMORY_ONLY_SER:‌ Similar to MEMORY_ONLY, but the Java objects are serialized. This should take less space, but reading will consume more CPU.

I write this simple program in order to test persist() functionality in dataframe:

SparkSession spark = SparkSession.builder()
    .appName("test persist")
    .master("local[*]")
    .getOrCreate();
String filename = "data/numbers.csv";
StructType schema = new StructType(new StructField[]{
          new StructField("value", DataTypes.IntegerType, false, Metadata.empty())});

Dataset df = spark
                      .read()
                      .schema(schema)
                      .option("header", false)
                      .csv(filename);
df = df.persist(StorageLevel.MEMORY_ONLY());
df.count();
df = df.withColumn("value", col("value").plus(5))
        .withColumn("value", col("value").plus(6));

ExplainCommand explain = new ExplainCommand(df.queryExecution().logical(), ExplainMode.fromString("codegen"));
  for (InternalRow internalRow : spark.sessionState()
          .executePlan(explain)
          .executedPlan()
          .executeCollect()) {
      System.out.println(internalRow.getString(0));
  }

numbers.csv is a simple text file contains numbers:

first I set storage level to StorageLevel.MEMORY_ONLY() then StorageLevel.MEMORY_ONLY_SER() and examine the generated code for both states, THEY ARE THE SAME!

for StorageLevel.MEMORY_ONLY_SER() the generated code is:

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 int columnartorow_batchIdx_0;
/* 010 */   private org.apache.spark.sql.execution.vectorized.OnHeapColumnVector[] columnartorow_mutableStateArray_2 = new org.apache.spark.sql.execution.vectorized.OnHeapColumnVector[1];
/* 011 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[] columnartorow_mutableStateArray_3 = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[2];
/* 012 */   private org.apache.spark.sql.vectorized.ColumnarBatch[] columnartorow_mutableStateArray_1 = new org.apache.spark.sql.vectorized.ColumnarBatch[1];
/* 013 */   private scala.collection.Iterator[] columnartorow_mutableStateArray_0 = new scala.collection.Iterator[1];
/* 014 */
/* 015 */   public GeneratedIteratorForCodegenStage1(Object[] references) {
/* 016 */     this.references = references;
/* 017 */   }
/* 018 */
/* 019 */   public void init(int index, scala.collection.Iterator[] inputs) {
/* 020 */     partitionIndex = index;
/* 021 */     this.inputs = inputs;
/* 022 */     columnartorow_mutableStateArray_0[0] = inputs[0];
/* 023 */
/* 024 */     columnartorow_mutableStateArray_3[0] = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(1, 0);
/* 025 */     columnartorow_mutableStateArray_3[1] = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(1, 0);
/* 026 */
/* 027 */   }
/* 028 */
/* 029 */   private void columnartorow_nextBatch_0() throws java.io.IOException {
/* 030 */     if (columnartorow_mutableStateArray_0[0].hasNext()) {
/* 031 */       columnartorow_mutableStateArray_1[0] = (org.apache.spark.sql.vectorized.ColumnarBatch)columnartorow_mutableStateArray_0[0].next();
/* 032 */       ((org.apache.spark.sql.execution.metric.SQLMetric) references[1] /* numInputBatches */).add(1);
/* 033 */       ((org.apache.spark.sql.execution.metric.SQLMetric) references[0] /* numOutputRows */).add(columnartorow_mutableStateArray_1[0].numRows());
/* 034 */       columnartorow_batchIdx_0 = 0;
/* 035 */       columnartorow_mutableStateArray_2[0] = (org.apache.spark.sql.execution.vectorized.OnHeapColumnVector) columnartorow_mutableStateArray_1[0].column(0);
/* 036 */
/* 037 */     }
/* 038 */   }
/* 039 */
/* 040 */   protected void processNext() throws java.io.IOException {
/* 041 */     if (columnartorow_mutableStateArray_1[0] == null) {
/* 042 */       columnartorow_nextBatch_0();
/* 043 */     }
/* 044 */     while ( columnartorow_mutableStateArray_1[0] != null) {
/* 045 */       int columnartorow_numRows_0 = columnartorow_mutableStateArray_1[0].numRows();
/* 046 */       int columnartorow_localEnd_0 = columnartorow_numRows_0 - columnartorow_batchIdx_0;
/* 047 */       for (int columnartorow_localIdx_0 = 0; columnartorow_localIdx_0 < columnartorow_localEnd_0; columnartorow_localIdx_0++) {
/* 048 */         int columnartorow_rowIdx_0 = columnartorow_batchIdx_0 + columnartorow_localIdx_0;
/* 049 */         boolean project_isNull_0 = true;
/* 050 */         int project_value_0 = -1;
/* 051 */         boolean columnartorow_isNull_0 = columnartorow_mutableStateArray_2[0].isNullAt(columnartorow_rowIdx_0);
/* 052 */         int columnartorow_value_0 = columnartorow_isNull_0 ? -1 : (columnartorow_mutableStateArray_2[0].getInt(columnartorow_rowIdx_0));
/* 053 */         if (!columnartorow_isNull_0) {
/* 054 */           project_isNull_0 = false; // resultCode could change nullability.
/* 055 */
/* 056 */           project_value_0 = columnartorow_value_0 + 11;
/* 057 */
/* 058 */         }
/* 059 */         columnartorow_mutableStateArray_3[1].reset();
/* 060 */
/* 061 */         columnartorow_mutableStateArray_3[1].zeroOutNullBytes();
/* 062 */
/* 063 */         if (project_isNull_0) {
/* 064 */           columnartorow_mutableStateArray_3[1].setNullAt(0);
/* 065 */         } else {
/* 066 */           columnartorow_mutableStateArray_3[1].write(0, project_value_0);
/* 067 */         }
/* 068 */         append((columnartorow_mutableStateArray_3[1].getRow()));
/* 069 */         if (shouldStop()) { columnartorow_batchIdx_0 = columnartorow_rowIdx_0 + 1; return; }
/* 070 */       }
/* 071 */       columnartorow_batchIdx_0 = columnartorow_numRows_0;
/* 072 */       columnartorow_mutableStateArray_1[0] = null;
/* 073 */       columnartorow_nextBatch_0();
/* 074 */     }
/* 075 */   }
/* 076 */
/* 077 */ }

and for StorageLevel.MEMORY_ONLY() the generated code is:

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 int columnartorow_batchIdx_0;
/* 010 */   private org.apache.spark.sql.execution.vectorized.OnHeapColumnVector[] columnartorow_mutableStateArray_2 = new org.apache.spark.sql.execution.vectorized.OnHeapColumnVector[1];
/* 011 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[] columnartorow_mutableStateArray_3 = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[2];
/* 012 */   private org.apache.spark.sql.vectorized.ColumnarBatch[] columnartorow_mutableStateArray_1 = new org.apache.spark.sql.vectorized.ColumnarBatch[1];
/* 013 */   private scala.collection.Iterator[] columnartorow_mutableStateArray_0 = new scala.collection.Iterator[1];
/* 014 */
/* 015 */   public GeneratedIteratorForCodegenStage1(Object[] references) {
/* 016 */     this.references = references;
/* 017 */   }
/* 018 */
/* 019 */   public void init(int index, scala.collection.Iterator[] inputs) {
/* 020 */     partitionIndex = index;
/* 021 */     this.inputs = inputs;
/* 022 */     columnartorow_mutableStateArray_0[0] = inputs[0];
/* 023 */
/* 024 */     columnartorow_mutableStateArray_3[0] = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(1, 0);
/* 025 */     columnartorow_mutableStateArray_3[1] = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(1, 0);
/* 026 */
/* 027 */   }
/* 028 */
/* 029 */   private void columnartorow_nextBatch_0() throws java.io.IOException {
/* 030 */     if (columnartorow_mutableStateArray_0[0].hasNext()) {
/* 031 */       columnartorow_mutableStateArray_1[0] = (org.apache.spark.sql.vectorized.ColumnarBatch)columnartorow_mutableStateArray_0[0].next();
/* 032 */       ((org.apache.spark.sql.execution.metric.SQLMetric) references[1] /* numInputBatches */).add(1);
/* 033 */       ((org.apache.spark.sql.execution.metric.SQLMetric) references[0] /* numOutputRows */).add(columnartorow_mutableStateArray_1[0].numRows());
/* 034 */       columnartorow_batchIdx_0 = 0;
/* 035 */       columnartorow_mutableStateArray_2[0] = (org.apache.spark.sql.execution.vectorized.OnHeapColumnVector) columnartorow_mutableStateArray_1[0].column(0);
/* 036 */
/* 037 */     }
/* 038 */   }
/* 039 */
/* 040 */   protected void processNext() throws java.io.IOException {
/* 041 */     if (columnartorow_mutableStateArray_1[0] == null) {
/* 042 */       columnartorow_nextBatch_0();
/* 043 */     }
/* 044 */     while ( columnartorow_mutableStateArray_1[0] != null) {
/* 045 */       int columnartorow_numRows_0 = columnartorow_mutableStateArray_1[0].numRows();
/* 046 */       int columnartorow_localEnd_0 = columnartorow_numRows_0 - columnartorow_batchIdx_0;
/* 047 */       for (int columnartorow_localIdx_0 = 0; columnartorow_localIdx_0 < columnartorow_localEnd_0; columnartorow_localIdx_0++) {
/* 048 */         int columnartorow_rowIdx_0 = columnartorow_batchIdx_0 + columnartorow_localIdx_0;
/* 049 */         boolean project_isNull_0 = true;
/* 050 */         int project_value_0 = -1;
/* 051 */         boolean columnartorow_isNull_0 = columnartorow_mutableStateArray_2[0].isNullAt(columnartorow_rowIdx_0);
/* 052 */         int columnartorow_value_0 = columnartorow_isNull_0 ? -1 : (columnartorow_mutableStateArray_2[0].getInt(columnartorow_rowIdx_0));
/* 053 */         if (!columnartorow_isNull_0) {
/* 054 */           project_isNull_0 = false; // resultCode could change nullability.
/* 055 */
/* 056 */           project_value_0 = columnartorow_value_0 + 11;
/* 057 */
/* 058 */         }
/* 059 */         columnartorow_mutableStateArray_3[1].reset();
/* 060 */
/* 061 */         columnartorow_mutableStateArray_3[1].zeroOutNullBytes();
/* 062 */
/* 063 */         if (project_isNull_0) {
/* 064 */           columnartorow_mutableStateArray_3[1].setNullAt(0);
/* 065 */         } else {
/* 066 */           columnartorow_mutableStateArray_3[1].write(0, project_value_0);
/* 067 */         }
/* 068 */         append((columnartorow_mutableStateArray_3[1].getRow()));
/* 069 */         if (shouldStop()) { columnartorow_batchIdx_0 = columnartorow_rowIdx_0 + 1; return; }
/* 070 */       }
/* 071 */       columnartorow_batchIdx_0 = columnartorow_numRows_0;
/* 072 */       columnartorow_mutableStateArray_1[0] = null;
/* 073 */       columnartorow_nextBatch_0();
/* 074 */     }
/* 075 */   }
/* 076 */
/* 077 */ }

one thing I can't understand is what is actually serialized or de-serialized when we call persist() on dataframe? isn't dataframe serialized already?

when we call persist on dataframe and specify storage level to be MEMORY_ONLY or MEMORY_ONLY_SER what is actually serialized/de-serialized?

0 Answers