下面是一段创建structured streaming的Dataset的代码：
val lines = spark.readStream.format("socket")
.option("host", "localhost").option("port", 9999).load();
会创建一个socket类型的Source，该name2class的映射由DataSource.lookupDataSource()完成
val serviceLoader = ServiceLoader.load(classOf[DataSourceRegister], loader)
...
serviceLoader.asScala.filter(_.shortName().equalsIgnoreCase(provider1)).toList
...
应该是从当前类路径中查找所有的DataSourceRegister，并读取它的shortName()，如果是"socket"就确定了由该DataSourceRegister来创建对应的DataSource
果然，有一个TextSocketSourceProvider
class TextSocketSourceProvider extends StreamSourceProvider with DataSourceRegister with Logging {
...
override def shortName(): String = "socket"
override def createSource(
sqlContext: SQLContext,
metadataPath: String,
schema: Option[StructType],
providerName: String,
parameters: Map[String, String]): Source = {
val host = parameters("host")
val port = parameters("port").toInt
new TextSocketSource(host, port, parseIncludeTimestamp(parameters), sqlContext)
}
}
TextSocketSourceProvider的createSource创建一个TextSocketSource
TextSocketSource是一个Source，Source接口如下：
trait Source {
def schema: StructType
def getOffset: Option[Offset]
def getBatch(start: Option[Offset], end: Offset): DataFrame
def commit(end: Offset) : Unit = {}
def stop(): Unit
}
