作为此注解的后续，在clojure中使用泛型类的类型提示
对于最新的flink版本（在1.6.1上测试），您需要定义一个自定义类，否则会出现如下错误：

Exception in thread "main" java.lang.IllegalArgumentException: No matching method found: returns for class org.apache.flink.api.java.operators.FlatMapOperator, compiling:(WordCount.clj:69:13)

自定义类：

package org.apache.flink.java;

import org.apache.flink.api.java.tuple.Tuple2;

public class WordCountTuple extends Tuple2<String, Integer> {

}

clojure代码

(ns org.apache.flink.clojure.WordCount
  (:import
   (org.apache.flink.api.common.functions FlatMapFunction)
   (org.apache.flink.api.java DataSet)
   (org.apache.flink.api.java ExecutionEnvironment)
   (org.apache.flink.api.java.tuple Tuple2)
   (org.apache.flink.java WordCountTuple)
   (org.apache.flink.util Collector)
   (java.lang String))
  (:require [clojure.string :as str])
  (:gen-class))

(def flink-env (ExecutionEnvironment/getExecutionEnvironment))

(def text (.fromElements flink-env (to-array ["please test me and me too"])))

(deftype tokenizer [] FlatMapFunction
         (flatMap [this value collector]
           (doseq [v (str/split value #"\s")]
             (.collect collector (Tuple2. v (int 1))))))

(def tokens (.returns (.flatMap text (tokenizer.)) WordCountTuple))

(def counts (.sum (.groupBy tokens (int-array [0])) 1))

(defn -main []
  (.print counts))

这里是工作示例https://github.com/guillaume/flink-external

clojure无法处理反射，因此需要通过flink方法手动指定返回类型 returns .

(.returns (.flatMap dataset (tokenizer)) String)

此外，您需要使用 deftype 定义 tokenizer 并在使用时示例化一个新对象，因为flink无法处理匿名类：

(deftype tokenizer [] FlatMapFunction
                      (flatMap [this value collector] 
                        (println value)))

(.flatMap dataset (tokenizer.))

下面是一个完整的“字数计算示例”，可以打包到jar中执行。
注意类型提示和类型转换。为了 tokenizer 输出 (int 1) 是必需的，否则 Long 会是第二种 Tuple2 . 此外，我们使用一个字符串来声明 tokenizer （类类型是不够的，因为还必须指定反射类型）。最后，我们需要输入提示 (int-array [0]) 以解决 groupBy （没有它，这个方法对clojure编译器来说是不明确的）。

(ns org.apache.flink.flink-clojure.WordCount
 (:import
 (org.apache.flink.api.common.functions FlatMapFunction)
 (org.apache.flink.api.java DataSet)
 (org.apache.flink.api.java ExecutionEnvironment)
 (org.apache.flink.api.java.tuple Tuple2)
 (org.apache.flink.util Collector)
 (java.lang String))
 (:require [clojure.string :as str])
 (:gen-class))

(def flink-env (ExecutionEnvironment/createLocalEnvironment))

(def text (.fromElements flink-env (to-array ["please test me and me too"])))

(deftype tokenizer [] FlatMapFunction
                      (flatMap [this value collector]
                        (doseq [v (str/split value #"\s")]
                          (.collect collector (Tuple2. v (int 1))))))

(def tokens (.returns (.flatMap text (tokenizer.)) "Tuple2<String,Integer>"))

(def counts (.sum (.groupBy tokens (int-array [0])) 1))

(defn -main []
  (.print counts)
)