flink 读取kafka 数据写入Neo4j

<dependencies>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>1.10.0</version>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_2.11</artifactId>
    <version>1.10.0</version>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.10_2.11</artifactId>
    <version>1.10.0</version>
  </dependency>
  <dependency>
    <groupId>org.neo4j</groupId>
    <artifactId>neo4j</artifactId>
    <version>3.5.3</version>
  </dependency>
  <dependency>
     <groupId>org.neo4j.driver</groupId>
     <artifactId>neo4j-java-driver</artifactId>
     <version>${neo4j.driver.version}</version>
  </dependency>
  <dependency>
     <groupId>com.alibaba</groupId>
     <artifactId>fastjson</artifactId>
     <version>1.2.66</version>
  </dependency>
  <dependency>
     <groupId>junit</groupId>
     <artifactId>junit</artifactId>
     <version>4.12</version>
  </dependency>
  <dependency>
     <groupId>org.json</groupId>
     <artifactId>json</artifactId>
     <version>20090211</version>
  </dependency>
</dependencies>

public class HashMapHelper <K,T> extends HashMap<K,T>{
    @Override
    public String toString() {
        StringBuffer sbf= new StringBuffer("{");
        for (Entry<K,T> entry : entrySet())
        {
           sbf.append(entry.getKey() + ":" + entry.getValue()+ ",");
        }
        sbf.setCharAt(sbf.length()-1,'}');
        return sbf.toString() ;
    }
}

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties prop = new Properties();
        prop.put("bootstrap.servers", broker);
        prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        prop.put("subscribe","db01_topic");
        prop.put("auto.offset.reset", "latest");
        prop.put("max.poll.records",2000);
        env.enableCheckpointing(5000);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        FlinkKafkaConsumer011<String> consumer = new FlinkKafkaConsumer011<>(topicId, new SimpleStringSchema(), prop);
        consumer.setStartFromLatest();
        DataStreamSource<String> dataStreamSource = env.addSource(consumer).setParallelism(2);
        /**
         * 解析json
//         */
       SingleOutputStreamOperator<String> jsonSource = dataStreamSource.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                HashMapHelper<String,String> map1 = new HashMapHelper<String,String>();
                JSONObject jsonData = new JSONObject(value);
                    JSONArray JSONArray = jsonData.getJSONArray("change");
                if(JSONArray!=null && JSONArray.length()>0) {
                    String valueData = JSONArray.get(0).toString();
                    JSONObject jsb = new JSONObject(valueData);
                    if ("insert".equals(jsb.get("kind"))) {
                        JSONArray columnnames = jsb.getJSONArray("columnnames");
                        JSONArray columnvalues = jsb.getJSONArray("columnvalues");
                        JSONArray columntypes = jsb.getJSONArray("columntypes");
                        for (int i = 0; i <= columnnames.length() - 1; i++) {
                            if (columntypes.get(i).toString().contains("geometry")) {
                                map1.put(columnnames.get(i).toString(), columnvalues.get(i).toString());
                            } else {
                                map1.put(columnnames.get(i).toString(), columnvalues.get(i).toString() + ";");
                            }
                        }
                    }
                }
                return map1.toString();
            }
        });

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.neo4j.driver.v1.*;
import java.util.Random;
public class Neo4jSink<T> extends RichSinkFunction<T> {
    private Driver driver;
    public static final String username = "neo4j";
    public static final String password =  "zhanyang";
    public static final String url = "bolt://192.168.32.102:7687";
    public Driver getDriver() {
        driver = GraphDatabase.driver(url, AuthTokens.basic(username, password), Config.build().toConfig());
        if(driver == null){
            System.out.println("****************A driver can`t be created to {}"+url);
        }
        return driver;
    }
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
          driver = getDriver();
    }
    /**
     *     run("MATCH (n) WHERE n.name = {myNameParam} RETURN (n)",
     *      * Values.parameters( "myNameParam", "Bob" ))
     */
    @Override
    public void invoke(T line) throws Exception {
        Session session = driver.session();
        String cyper ="MERGE (tuple2:Tuple {id:"+new Random().nextInt()+",";
        for (String elems : line.toString().split(",", -1)) {
            String key = StringUtils.isBlank(elems.split(":")[0])   ? "null" :elems.split(":")[0].replace("{","");
            String value = StringUtils.isBlank(elems.split(":")[1])  ? "null" :elems.split(":")[1].replace("}","");
            cyper=cyper+key+":'"+value+"',";
        }
        cyper=cyper.substring(0,cyper.length()-1)+"}) return tuple2";
        System.out.println("*******"+cyper);
        session.run(cyper);
    }
    @Override
    public void close() throws Exception {
        ....
    }
}

jsonSource.addSink(new Neo4jSink<>());
env.execute("Flink add kafka DataSource");