[SPARK-4361][Doc] Add more docs for Hadoop Configuration

[zsxwing]

I'm trying to point out reusing a Configuration in these APIs is dangerous. Any better idea?

[SparkQA]

Test build #23263 has started for PR 3225 at commit fe4e3d5.

• This patch merges cleanly.

[SparkQA]

Test build #23263 has finished for PR 3225 at commit fe4e3d5.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/23263/
Test PASSed.

[rxin]

i don't think we reuse the conf across different RDDs, do we?

[zsxwing]

People may call this method directly and pass their Configuration.

   def newAPIHadoopFile[K, V, F <: NewInputFormat[K, V]](
       path: String,
       fClass: Class[F],
       kClass: Class[K],
       vClass: Class[V],
       conf: Configuration = hadoopConfiguration)

E.g., creating a configuration for accessing hbase:

import java.io.{DataOutputStream, ByteArrayOutputStream}
import java.lang.String
import org.apache.hadoop.hbase.client.Scan
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Base64

def convertScanToString(scan: Scan): String = {
  val out: ByteArrayOutputStream = new ByteArrayOutputStream
  val dos: DataOutputStream = new DataOutputStream(out)
  scan.write(dos)
  Base64.encodeBytes(out.toByteArray)
}

val conf = HBaseConfiguration.create()
val scan = new Scan()
scan.setCaching(500)
scan.setCacheBlocks(false)
conf.set(TableInputFormat.INPUT_TABLE, "table_name")
conf.set(TableInputFormat.SCAN, convertScanToString(scan))
val rdd = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])
rdd.count()

This is fine. However, some people may need to access two tables and union them. They may reuse the Configuration like this:

val conf = HBaseConfiguration.create()
val scan = new Scan()
scan.setCaching(500)
scan.setCacheBlocks(false)
conf.set(TableInputFormat.INPUT_TABLE, "table_name")
conf.set(TableInputFormat.SCAN, convertScanToString(scan))
val rdd = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])

conf.set(TableInputFormat.INPUT_TABLE, "another_table_name")
val rdd2 = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])

rdd.union(rdd2).count()

The result will be weird and wrong.
I think the docs should tell people not to reuse it like this.

My motivation is this mail thread: http://apache-spark-user-list.1001560.n3.nabble.com/How-did-the-RDD-union-work-td18686.html

[JoshRosen]

In fact, there are many uses of sc.hadoopConfiguration in the wild which assume that it's shared:

https://github.com/search?q=sc.hadoopConfiguration&type=Code&utf8=%E2%9C%93

Most of those are using it to configure S3 credentials.

[andrewor14]

Nice find. It seems perfectly reasonable from the user's perspective to just save sc.hadoopConfiguration into a val and use it for many things. That's probably what I would have done if I didn't know about the nuances here.

[andrewor14]

@JoshRosen

[JoshRosen]

I agree that the current mutable nature of sc.hadoopConfiguration is confusing and this seems like it's worth documenting. It would be nicer if we didn't have this messy mutable configuration, though. I think that the combination of a mutable conf + lazy evaluation is what makes this confusing, since @zsxwing's example of reading from two tables would work correctly under eager evaluation:

conf.set(TableInputFormat.INPUT_TABLE, "table_name")
conf.set(TableInputFormat.SCAN, convertScanToString(scan))
val rdd = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])

conf.set(TableInputFormat.INPUT_TABLE, "another_table_name")
val rdd2 = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])

I suppose one approach would be to have the configuration stay mutable but to make a defensive copy of it when constructing RDDs that accept configurations. This would break programs that were relying on being able to mutate credentials after having defined a bunch of RDDs (e.g. define some RDDs, fail due to missing S3 credentials, supply new credentials, and re-run), but I think it makes things easier to reason about.

If we're not going to introduce any change in behavior, though, then I think we should document the current behavior more explicitly, as this patch has done.

[srowen]

Is the outcome here that the doc changes are OK to merge? documenting current behavior seems good.

[andrewor14]

really small nit but this should be javadoc style instead of scaladoc

[andrewor14]

I think it's safe to say that we won't implement the alternative behavior that @JoshRosen suggested by the release. For this reason I think we should at least document these unexpected behavior for 1.3 in addition to delaying the fix till later. I'm going to merge this into master and 1.3.