代码混用
 
在java代码编写代码时候不知不觉就引入了，引入了scala包，
 
  <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
           <version>${flink.version}</version>
       </dependency>
       
 
package cn.putact.datastream;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
**import scala.Tuple2;**

public class NcWordCount {
    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<String, Integer>> dataStream = env
                .socketTextStream("localhost", 9999)
                .flatMap(new Splitters())
                .keyBy(value -> value._1)
                .timeWindow(Time.seconds(5))
                .sum(1);
        dataStream.print();

        env.execute("Window WordCount");
    }

    public static class Splitters implements FlatMapFunction<String,Tuple2<String,Integer>>{
        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {

            for (String word:s.split(" ")){
                collector.collect(new Tuple2<String ,Integer>(word,1));
            }
        }
    }
}


 
Exception in thread "main" org.apache.flink.api.common.typeutils.CompositeType$InvalidFieldReferenceException: Cannot reference field by position on GenericType<scala.Tuple2>Referencing a field by position is supported on tuples, case classes, and arrays. Additionally, you can select the 0th field of a primitive/basic type (e.g. int).
	at org.apache.flink.streaming.util.typeutils.FieldAccessorFactory.getAccessor(FieldAccessorFactory.java:87)
	at org.apache.flink.streaming.api.functions.aggregation.SumAggregator.<init>(SumAggregator.java:43)
	at org.apache.flink.streaming.api.datastream.WindowedStream.sum(WindowedStream.java:1352)
	at cn.putact.datastream.NcWordCount.main(NcWordCount.java:21)
 
调试了半天最后发现java代码混合了scala报错，idea编译是通过的，最好写纯java代码，mavn依赖去掉scala 包，代码中用java tuple元祖就好
 
//import scala.Tuple2;
import org.apache.flink.api.java.tuple.Tuple2;

Tuple2 报错Incompatible equality constrain T1 and String
 
 
**解决方式：**可能是spark_core依赖包版本问题，更换一个版本就行了。

使用UntypedAbstractActor时，出现message匹配不上自定义的对象，但是debug时发现类型确实是，而且用了case class，各种方法都不行
 
类结构，最终有效代码：
 
abstract class Master(latch: CountDownLatch) extends UntypedAbstractActor {
override def onReceive(message: Any) = {

    logger.info("master,actor is:" + self)
    logger.info("Master当前消息类型：" + message.getClass.getSimpleName)
    message match {
        //启动
        case start: String => {
            logger.info("=======================start================")
            visitedPageStore.add(start)
            getParser() ! visitedPageStore.getNext()
        }
        //页面
        case (content: PageContent, _) => {
            logger.info("========================page==============")
            getIndexer() ! content
            //存储待访问页面链接
            visitedPageStore.addAll(content.getLinksToFollow())
            if (visitedPageStore.isFinished()) {
                //完成了则提交
                getIndexer() ! COMMIT_MESSAGE
            } else {
                //继续获取下一个页面
                for (page <- visitedPageStore.getNextBatch()) {
                    getParser() ! page
                }
            }

        }
        //索引
        case (indexedMessage: IndexedMessage, _) => {
            logger.info("====================index=================")
            visitedPageStore.finished(indexedMessage.getPath)
            if (visitedPageStore.isFinished())
                getIndexer() ! COMMIT_MESSAGE

        }
        //提交
        case (COMMITTED_MESSAGE, _) => {
            logger.info("======================end================")
            logger.info("Shutting down, finished")
            getContext().system.terminate()
            latch.countDown()
        }
        case _ => {
            logger.info("Unknown execution steps")
        }
    }

}
}
 
子类：
class SimpleActorMaster(latch: CountDownLatch) extends Master(latch) {}
 
打印当前message的类型输出如下：
 
Master当前消息类型：String
 
Master当前消息类型：Tuple2
 
这里第二次匹配case的时候竟然是元组类型，仔细看debug，
 
难怪原来使用case content: PageContent一直匹配不成功。。原来message消息类型不是纯粹的就是消息本身的类型，可能还携带了actor地址。。。
 
 

今天在写spark 提数的时候遇到一个异常，如下
 
Caused by: java.lang.ClassCastException: org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema cannot be cast to scala.Tuple2
  at $anonfun$1$$anonfun$apply$1.apply(<console>:27)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
  at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
  at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35)
  at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
  at scala.collection.AbstractTraversable.map(Traversable.scala:104)
  at $anonfun$1.apply(<console>:27)
  at $anonfun$1.apply(<console>:27)
  at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
  at scala.collection.Iterator$class.foreach(Iterator.scala:893)
  at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
  at org.apache.spark.rdd.RDD$$anonfun$foreach$1$$anonfun$apply$28.apply(RDD.scala:917)
  at org.apache.spark.rdd.RDD$$anonfun$foreach$1$$anonfun$apply$28.apply(RDD.scala:917)
  at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
  at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
  at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
  at org.apache.spark.scheduler.Task.run(Task.scala:99)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
  at java.lang.Thread.run(Thread.java:745)
 
df的schema如下：
 
root
 |-- cityId: long (nullable = true)
 |-- countryId: long (nullable = true)
 |-- outline: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- _1: double (nullable = true)
 |    |    |-- _2: double (nullable = true)
 |-- provinceId: long (nullable = true)
 |-- townId: long (nullable = true)
 
使用如下方法在提取outline 字段时，报错
 
val outline = df.select("outline").collect().map(row => row.getAs[Seq[(Double,Double)]]("outline"))
 
解决方法，我已知两种： 
 第一种，在生成数据时，定义case class对象，不存储tuple，就我个人而言，不适用，数据已经生成完毕，再次生成数据需要2天的时间 
 第二种方法：
 
 val lines = json.filter(row => !row.isNullAt(2)).select("outline").rdd.map(r => {
      val row:Seq[(Double,Double)] = r.getAs[Seq[Row]](0).map(x =>{(x.getDouble(0),x.getDouble(1))})
      row
    }).collect()(0)
 
至于报错的原因，google了一下，我觉得有一种说法可信： 
 在row 中的这些列取得时候，要根据类型取，简单的像String，Seq[Double] 这种类型就可以直接取出来，但是像 Seq[(Double,Double)] 这种类型直接取得花就会丢失schema信息，虽然值能取到，但是schema信息丢了，在dataFrame中操作的时候就会抛错