java.lang.UnsupportedOperationException) empty.min java.lang.UnsupportedOperationException: empty.min at scala.collection.TraversableOnce.min...(TraversableOnce.scala:225) at scala.collection.TraversableOnce.min$(TraversableOnce.scala:223) at
---- 过滤 filter 过滤符合一定条件的元素 定义 方法签名 def filter(p: (A) ⇒ Boolean): TraversableOnce[A] 方法解析 filter方法...API 说明 参数 p: (A) ⇒ Boolean 传入一个函数对象接收一个集合类型的参数返回布尔类型，满足条件返回true, 不满足返回false 返回值 TraversableOnce
与之前的不同，这里的Lambda表达式输入是泛型T，输出是一个TraversableOnce[R],TraversableOnce表示这是一个R组成的列表。...使用TraversableOnce也导致我们无论如何都要返回一个列表，即使是一个空列表，否则无法匹配函数的定义。...的句子进行处理 val longSentenceWords = dataStream.flatMap { input => { if (input.size > 15) { // 输出是 TraversableOnce...这里将Array[String]转成了Seq[String] input.split(" ").toSeq } else { // 为空时必须返回空列表，否则返回值无法与TraversableOnce...比如在本例中，如果返回值不是一个TraversableOnce，那么Intellij Idea会将该行标红，告知我们输入或输出的类型不匹配。
$$anonfun$foldLeft$1 (scala.collection) apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection...$$anonfun$foldLeft$1 (scala.collection) apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection...$$anonfun$foldLeft$1 (scala.collection) apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection...$$anonfun$foldLeft$1 (scala.collection) apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection...$$anonfun$foldLeft$1 (scala.collection) apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection
org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram$$anonfun$optimize$1.apply(FlinkChainedProgram.scala:58) 在 scala.collection.TraversableOnce...$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157) 在 scala.collection.TraversableOnce$$anonfun...$foldLeft$1.apply(TraversableOnce.scala:157) 在 scala.collection.Iterator$class.foreach(Iterator.scala...在 scala.collection.AbstractIterable.foreach(Iterable.scala:54) 在 scala.collection.TraversableOnce...$class.foldLeft(TraversableOnce.scala:157) 在 scala.collection.AbstractTraversable.foldLeft(Traversable.scala
下面是新版本的jdbcQueryResult函数： def jdbcQueryResult[C[_] <: TraversableOnce[_], A]( ctx: JDBCQueryContext...}) } Source.fromPublisher[A](publisher) } def jdbcQueryResult[C[_] <: TraversableOnce...} } } private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext...Seq(result.toLong).to[C] } } } private def singleTxUpdate[C[_] <: TraversableOnce...} results.to[C] } } } def jdbcTxUpdates[C[_] <: TraversableOnce
将处理好的元素封装到新的列表中, 并返回 而foreach不会返回我们新的列表 所以一般视使用场景, 来选择带返回的map还是不返回的foreach 用法 方法签名 def map[B](f: (A) ⇒ B): TraversableOnce...所以可以自动推断) 参数 f: (A) ⇒ B 传入一个函数对象该函数接收一个元素A（要转换的列表元素），返回值为类型B 返回值 TraversableOnce[B] B类型的集合 traversable
批次更新batch-update是指多条输入参数在一条统一的statement上施用： def jdbcBatchUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext...无论单条或多条update函数又被分为返回主键或更新状态类型的函数： private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_...Seq(result.toLong).to[C] } } } private def singleTxUpdate[C[_] <: TraversableOnce...} } } private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext...Seq(result.toLong).to[C] } } } private def singleTxUpdate[C[_] <: TraversableOnce
可能大家还是有点晕, 我们向下看, 到下面具体的例子就会明白. flatten再将整个列表进行扁平化 方法签名 def flatMap[B](f: (A) ⇒ GenTraversableOnce[B]): TraversableOnce...泛型 [B] 最终要转换的集合元素类型 参数 f: (A) ⇒ GenTraversableOnce[B] 传入一个函数对象函数的参数是集合的元素函数的返回值是一个集合 返回值 TraversableOnce
"SQL_EXEDDL executed succesfully." } } } } def jdbcBatchUpdate[C[_] <: TraversableOnce...} } } private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(...usql.apply() Seq(result.toLong).to[C] } } } private def singleTxUpdate[C[_] <: TraversableOnce...} else true } def noActon: PreparedStatement=>Unit = pstm => {} def multiTxUpdates[C[_] <: TraversableOnce....toLong } } results.to[C] } } } def jdbcTxUpdates[C[_] <: TraversableOnce
._ def fetchResultPage[C[_] <: TraversableOnce[_],A](ctx: CQLQueryContext[A], pageSize: Int = 100)...(resultSet,(resultSet.asScala.view.map(ctx.extractor)).to[C]) } def fetchMorePages[C[_] <: TraversableOnce...} } } private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(...usql.apply() Seq(result.toLong).to[C] } } } private def singleTxUpdate[C[_] <: TraversableOnce....toLong } } results.to[C] } } } def jdbcTxUpdates[C[_] <: TraversableOnce
接着调用了flatMap方法： def flatMap[U: ClassTag](flatMapFunc: T => TraversableOnce[U]): DStream[U] = ssc.withScope...streaming] class FlatMappedDStream[T: ClassTag, U: ClassTag]( parent: DStream[T], flatMapFunc: T => TraversableOnce
方法解析 map方法 API 说明 泛型 [B] 指定map方法最终返回的集合泛型 参数 f: (A) ⇒ B 传入一个函数对象该函数接收一个类型A（要转换的列表元素），返回值为类型B 返回值 TraversableOnce...flatmap方法 API 说明 泛型 【B】 最终要转换的集合元素类型 参数 f: (A) ⇒ GenTraversableOnce[B] 传入一个函数对象函数的参数是集合的元素函数的返回值是一个集合 返回值 TraversableOnce...方法解析 filter方法 API 说明 参数 p: (A) ⇒ Boolean 传入一个函数对象接收一个集合类型的参数返回布尔类型，满足条件返回true, 不满足返回false 返回值 TraversableOnce
其中 f 本身是 BaseType => A 类型 2.6、flatMap[A](f: BaseType => TraversableOnce[A]): Seq[A] 原理与 map 一致，只是 f...变成了 BaseType => TraversableOnce[A] 2.5、collect[B](pf: PartialFunction[BaseType, B]): Seq[B] def collect
下面是一个例子，在Scala集合 trait TraversableOnce 定义了 foldLeft def foldLeft[B](z: B)(op: (B, A) => B): B foldLeft
} } } private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(...usql.apply() Seq(result.toLong).to[C] } } } private def singleTxUpdate[C[_] <: TraversableOnce....toLong } } results.to[C] } } } def jdbcTxUpdates[C[_] <: TraversableOnce...)) } object CQLEngine { import CQLContext._ import CQLHelpers._ def fetchResultPage[C[_] <: TraversableOnce...(resultSet,(resultSet.asScala.view.map(ctx.extractor)).to[C]) } def fetchMorePages[C[_] <: TraversableOnce
对每个元素进行变换，应用变换函数 //(T)=>V filter() //过滤器,(T)=>Boolean flatMap() //压扁,T => TraversableOnce
} } } private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(...usql.apply() Seq(result.toLong).to[C] } } } private def singleTxUpdate[C[_] <: TraversableOnce...} else true } def noActon: PreparedStatement=>Unit = pstm => {} def multiTxUpdates[C[_] <: TraversableOnce....toLong } } results.to[C] } } } def jdbcTxUpdates[C[_] <: TraversableOnce...{Await, Future} import scala.concurrent.duration._ def fetchResultPage[C[_] <: TraversableOnce[_
ResultSet execute(Statement statement); 返回结果ResultSet经过转换后成为scala collection： def fetchResultPage[C[_] <: TraversableOnce...use}). */ ListenableFuture fetchMoreResults(); 下面是分页持续读取的实现： def fetchMorePages[C[_] <: TraversableOnce...)) } object CQLEngine { import CQLContext._ import CQLHelpers._ def fetchResultPage[C[_] <: TraversableOnce...(resultSet,(resultSet.asScala.view.map(ctx.extractor)).to[C]) } def fetchMorePages[C[_] <: TraversableOnce
领取专属 10元无门槛券
手把手带您无忧上云
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储
