flink学习(1)读取文件并求和

今天尝试了flink读取文件的测试程序编写，代码比较简单，途中遇到了一个问题，自定义封装类型的时候编译抛出如下异常：

Error:(10, 20) could not find implicit value for evidence parameter of type org.apache.flink.api.common.typeinfo.TypeInformation[whTest.readFile.SexAndAge]
     dataStream.map(formatPerson(_)).print()

这个问题是因为程序需要一个隐式参数(implicit parameter)。我们可以看看上面程序用到的 map 在Flink中的实现：

def map[R: TypeInformation](fun: T => R): DataStream[R] = {
  if (fun == null) {
    throw new NullPointerException("Map function must not be null.")
  }
  val cleanFun = clean(fun)
  val mapper = new MapFunction[T, R] {
    def map(in: T): R = cleanFun(in)
  }
 
  map(mapper)
}

在 map 的定义中有个 [R: TypeInformation] ，但是我们程序并没有指定任何有关隐式参数的定义，这时候编译代码无法创建TypeInformation，所以出现上面提到的异常信息。解决这个问题有以下两种方法

（1）、我们可以直接在代码里面加上以下的代码：

implicit val typeInfo = TypeInformation.of(classOf[SexAndAge])
然后再去编译代码就不会出现上面的异常。

（2）、官方推荐的做法是在代码中引入一下包：

import org.apache.flink.streaming.api.scala._
对于静态数据集，我们可以引入以下包：

import org.apache.flink.api.scala._
问题就得到解决了

以下是测试的代码：

package whTest

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

/**
  * 导入 org.apache.flink.streaming.api.scala._
  * 解决异常 could not find implicit value for evidence parameter of type
  *异常的发生通常是因为程序需要一个隐式参数(implicit parameter)
  */
import org.apache.flink.streaming.api.scala._
object ReadFile {
//自行定义一个封装类型
  case class SexAndAge (sex:String,age:Int)
  def main(args: Array[String]): Unit = {
    val env : StreamExecutionEnvironment  = StreamExecutionEnvironment.getExecutionEnvironment
    val dataStream:DataStream[String]  = env.readTextFile("C:\\Users\\wh\\Desktop\\test.txt")
    val personDataStream = dataStream.map(formatPerson(_))
    //男女用户人数统计
    //如果要多多个字段求和可以使用reduce函数
    val counts = personDataStream.keyBy("sex").sum("age")
    counts.print()
    // execute program
    env.execute("Streaming WordCount")
  }
  def formatPerson(mess:String)={
    val split = mess.split(",")
    SexAndAge (split(1),split(2).toInt)
  }
}

输入文件如下：

wh,m,12
wc,m,11
xds,m,11
hr,f,11
cxy,f,11

测试结果：

3> SexAndAge(m,34)

1> SexAndAge(f,22)

最后编辑于：2019.05.10 15:31:32