import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.api.common.functions.FilterFunction;

public class Example {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Person> flintstones = env.fromElements(
                new Person("Fred", 35),
                new Person("Wilma", 35),
                new Person("Pebbles", 2));

        DataStream<Person> adults = flintstones.filter(new FilterFunction<Person>() {
            @Override
            public boolean filter(Person person) throws Exception {
                return person.age >= 18;
            }
        });

        adults.print();

        env.execute();
    }

    public static class Person {
        public String name;
        public Integer age;
        public Person() {}

        public Person(String name, Integer age) {
            this.name = name;
            this.age = age;
        }

        public String toString() {
            return this.name.toString() + ": age " + this.age.toString();
        }
    }
}

import org.apache.flink.streaming.api.scala._
import org.apache.flink.api.common.functions.FilterFunction

object Example {
  def main(args: Array[String]) {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val flintstones = env.fromElements(
      new Person("Fred", 35),
      new Person("Wilma", 35),
      new Person("Pebbles", 2))

    val adults = flintstones.filter(new FilterFunction[Person] {
      override def filter(person: Person): Boolean = {
        person.age >= 18
      }
    })

    adults.print()

    env.execute("Example")
  }

  class Person(var name: String, var age: Int) {
    def this() {
      this("", 0)
    }

    override def toString: String = {
      name + ": age " + age.toString
    }
  }
}