字符串相等但行为不同：如果文件字符串来自HDFS文件，则它们不适用于火花

Question

这个问题更容易用代码内联解释：

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.fs.Path

val sqlcontext = new org.apache.spark.sql.SQLContext(sc)

// I have a file (fileToProcess) in HDFS that contains the name of another HDFS file:
val fs = FileSystem.get(new Configuration())
val fileToProcess = "hdfs://production/user/robin/fileToProcess"
// The contents of fileToProcess is just the name of another file.  In this case
// hdfs://production/user/robin/people
// I want to read hdfs://production/user/robin/people and use it in the data frame as coded below.
// However, if I do this by reading the HDFS file (fileToProcess) to get the file name like this:
val file = fs.open(new Path(fileToProcess)).readLine().toString

// I will get a Task not serializable exception on the last line of the script

// If I hardcode the file name like this:
val file2 = "hdfs://production/user/robin/people"
// It works great; however, I can't do this as I don't know the file I need to read in reality
// file2 and file are both Strings seem equal in every way so I am really perplexed!

// Here is what I am doing with the file to get the exception

// The contents of people:
// { "person" : "John"}
// { "person" : "Sue"}
// { "person" : "Jean"}
// { "person" : "Jane"}
// { "person" : "John"}

val df = sqlcontext.read.json(file)
val peopleList = df.map(r => r(0).toString).distinct.collect
val anotherList = sc.parallelize(Array("Jean", "Sue", "Bill"))

val peopleListBroadcast = sc.broadcast(peopleList)

// everything works great up to this point

val filteredPeople = anotherList.filter(x=> peopleListBroadcast.value contains x)
// here I get a Task not serializable exception if I use the file name read from the HDFS file but it works fine if I hardcode it (like with file2)

我已经被困在这个舞台问题上好几天了。我似乎也找不到工作。有我看不到的差异吗？两个相等的字符串怎么会表现得如此不同呢。请帮帮我，我要发疯了，想办法解决这个问题！

我得到的具体例外是：

原因：java.io.NotSerializableException：org.apache.hadoop.hdfs.分布式文件系统序列化堆栈：-对象不可序列化（类：org.apache.hadoop.hdfs.分布式文件系统，值：DFS[DFSClient[clientName=DFSClient_NONMAPREDUCE_1011603383_1，ugi=robin（身份：SIMTER)]]) - 字段（类：$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$iwC$$iwC$iwC，名称：fs，类型：类org.apache.hadoop.fs.文件系统）

顺便说一句，我正在使用Spark 1.6.1和Scala 2.10.5。我认为任何人都应该能够通过在hdfs中制作这两个文件，然后将上面的代码放在spack-shell中来重新创建它

谢谢，罗宾

Answer 1

另一种选择是

fs = sc.textFile(fileToProcess,1).collect()

这将为您提供文件名

Answer 2

它与字符串无关。您将org的一个实例放入范围。阿帕奇。hadoop。财政司司长。文件系统不可序列化。将其标记为瞬态应能解决此特定问题：

@transient val fs = FileSystem.get(new Configuration())