javaFlink

source读取文件

//第二个是编码格式 
Iterator lines = Source.fromFile(new File("input/click.txt"),"utf8").getLines();
        while (lines.hasNext()){
            String next = lines.next();
            System.out.println(next);
        }

获取当前时间戳

import java.util.Calendar;

long l = Calendar.getInstance().getTimeInMillis();

source mysql

package p33;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;


public class RClickSOurce extends RichSourceFunction {
    Connection conn =null;
    PreparedStatement ps =null;
    @Override
    public void open(Configuration parameters) throws Exception {
       Class.forName("com.mysql.jdbc.Driver");
        conn=DriverManager.getConnection("jdbc:mysql://hdp1:3306/1912b", "root", "root");
         ps=conn.prepareStatement("select  * from t_event");
    }

    @Override
    public void close() throws Exception {
        conn.close();
        ps.close();
    }

    @Override
    public void run(SourceContext ctx) throws Exception {
        ResultSet resultSet = ps.executeQuery();
        while (resultSet.next()){
            String users = resultSet.getString(1);
            String uils = resultSet.getString(2);
            Long ts = resultSet.getLong(3);
            ctx.collect(new Event(users,uils,ts));
        }
    }

    @Override
    public void cancel() {

    }
}

maxby 求最大 里面的值只能写字段名

 KeyedStream keyedStream = eventSource.keyBy(data -> data.user);
        SingleOutputStreamOperator max = keyedStream.maxBy("ts");

reduce的规约聚合

package p42;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import p33.Event;


public class TranformReduceTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource eventSource = env.fromElements(new Event("bob", "./home", 1000L),
                new Event("mary", "./home", 2000L),
                new Event("bob", "./pord", 3000L)
        );
       //统计每个项目访问次数
        SingleOutputStreamOperator> map = eventSource.map(new MapFunction>() {
            @Override
            public Tuple2 map(Event event) throws Exception {
                return Tuple2.of(event.user, 1L);
            }
        });
        KeyedStream, String> keyedStream = map.keyBy(new KeySelector, String>() {
            @Override
            public String getKey(Tuple2 v1) throws Exception {
                return v1.f0;
            }
        });
        SingleOutputStreamOperator> reduce = keyedStream.reduce(new ReduceFunction>() {
            @Override
            public Tuple2 reduce(Tuple2 v1, Tuple2 t1) throws Exception {
                return Tuple2.of(v1.f0, v1.f1 + t1.f1);
            }
        });
        //选取当前最活跃的用户
        //这个key 是随便写的  用于把短的数据分到一个组
        SingleOutputStreamOperator> result = reduce.keyBy(data -> "key").reduce(new ReduceFunction>() {
            @Override
            public Tuple2 reduce(Tuple2 v1, Tuple2 t1) throws Exception {
                return v1.f1 > t1.f1 ? v1 : t1;
            }
        });
        result.print();
        env.execute();
    }
}

拿window 的ip地址

ipconfig

192.168.0.105

本地改localhost