code: def from_rdd_of_dataframes(self, rdd, column_idxs=None):
          
          def frame_to_spark_sql(frame):
              """Convert a Panda's DataFrame into Spark SQL Rows"""
              return [r.tolist() for r in frame.to_records()]
  
          def frame_to_schema_and_idx_names(frames):
              """Returns the schema and index names of the frames. Useful
              if the frame is large and we wish to avoid transfering
              the entire frame. Only bothers to apply once per partiton"""
              try:
                  frame = frames.next()
                  return [(list(frame.columns), list(frame.index.names))]
              except StopIteration:
                  return []
  
          # Store if the RDD was persisted so we don't uncache an
          # explicitly cached input.
          was_persisted = rdd.is_cached
          # If we haven't been supplied with the schema info cache the RDD
          # since we are going to eveluate the first partition and then eveluate
          # the entire RDD as part of creating a Spark DataFrame.
          (schema, index_names) = ([], [])
          if not column_idxs:
              rdd.cache()
              (schema, index_names) = rdd.mapPartitions(
                  frame_to_schema_and_idx_names).first()
          else:
              (schema, index_names) = column_idxs
          # Add the index_names to the schema.
          index_names = _normalize_index_names(index_names)
          schema = index_names + schema
          ddf = DataFrame.from_schema_rdd(
              self.sql_ctx.createDataFrame(rdd.flatMap(frame_to_spark_sql),
                                           schema=schema))
          ddf._index_names = index_names
          if not was_persisted:
              rdd.unpersist()
          return ddf
docs: Take an RDD of Panda's DataFrames and return a Dataframe.
          If the columns and indexes are already known (e.g. applyMap)
          then supplying them with columnsIndexes will skip eveluating
          the first partition to determine index info.