我正在尝试使用ApacheFlink通过twitter流api获取一些消息。
但是，我的代码没有在输出文件中写入任何内容。我正在计算特定单词的输入数据。
请看我的例子：

import java.util.Properties

import org.apache.flink.api.scala._
import org.apache.flink.streaming.connectors.twitter._
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import com.twitter.hbc.core.endpoint.{Location, StatusesFilterEndpoint, StreamingEndpoint}
import org.apache.flink.streaming.api.windowing.time.Time

import scala.collection.JavaConverters._

//////////////////////////////////////////////////////
// Create an Endpoint to Track our terms
class myFilterEndpoint extends TwitterSource.EndpointInitializer with Serializable {
  @Override
  def createEndpoint(): StreamingEndpoint = {
    //val chicago = new Location(new Location.Coordinate(-86.0, 41.0), new Location.Coordinate(-87.0, 42.0))
    val endpoint = new StatusesFilterEndpoint()
    //endpoint.locations(List(chicago).asJava)
    endpoint.trackTerms(List("odebrecht", "lava", "jato").asJava)
    endpoint
  }
}

object Connection {
  def main(args: Array[String]): Unit = {

    val props = new Properties()

    val params: ParameterTool = ParameterTool.fromArgs(args)
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.getConfig.setGlobalJobParameters(params)
    env.setParallelism(params.getInt("parallelism", 1))

    props.setProperty(TwitterSource.CONSUMER_KEY, params.get("consumer-key"))
    props.setProperty(TwitterSource.CONSUMER_SECRET, params.get("consumer-key"))
    props.setProperty(TwitterSource.TOKEN, params.get("token"))
    props.setProperty(TwitterSource.TOKEN_SECRET, params.get("token-secret"))

    val source = new TwitterSource(props)
    val epInit = new myFilterEndpoint()

    source.setCustomEndpointInitializer(epInit)

    val streamSource = env.addSource(source)

    streamSource.map(s => (0, 1))
      .keyBy(0)
      .timeWindow(Time.minutes(2), Time.seconds(30))
      .sum(1)
      .map(t => t._2)
      .writeAsText(params.get("output"))

    env.execute("Twitter Count")
  }
}

关键是，我没有错误信息，我可以看到在我的 Jmeter 板。我的消息源正在向我的triggerwindow发送数据。但它不能接收任何数据：

我一次有两个问题。
第一：如果没有收到任何信息，为什么我的源代码会向triggerwindow发送字节？
第二：我的代码有什么问题，我不能从twitter获取数据吗？