Java 8 stream and lambda function important examples

-

To Understand example please follow main method of StreamAPI class


import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

public class StreamAPI {

   public static void main(String[] args) {
      List list = new ArrayList<>();

      list.add(new CheckDomain(100l,”GOC1″,new Date()));
      list.add(new CheckDomain(100l,”GOC1″,new Date()));
      list.add(new CheckDomain(101l,”GOC2″,new Date()));
      list.add(new CheckDomain(101l,”GOC2″,new Date()));
      list.add(new CheckDomain(101l,”GOC2″,new Date()));     
      list.add(new CheckDomain(102l,”GOC3″,new Date()));
      list.add(new CheckDomain(103l,”GOC4″,new Date()));
      list.add(new CheckDomain(104l,”GOC5″,new Date()));
      list.add(new CheckDomain(105l,”GOC6″,new Date()));

      System.out.println(“list:”+list);

      //TO filter records using stream
      //Filtering out records whose run id is 101

      List filteredList = list.stream().filter( e-> e.getRunId()!=101l).collect(Collectors.toList());

      System.out.println(“filteredList:”+filteredList);

      // to fetch list of specific attribute from list of domains
      //Fetching all gocs
      List gocs = list.stream().map(CheckDomain::getGocDetails).collect(Collectors.toList());
      System.out.println(“gocs:”+gocs);

      // Fetching all unique run ids
      Set uniqueRunIds = list.stream().map(CheckDomain::getRunId).collect(Collectors.toSet());
      System.out.println(“uniqueRunIds:”+uniqueRunIds);

      // TO change value in list but this will update same value to all records so we will first filter and then will update
      list.stream().filter(e -> e.getGocDetails().equals(“GOC1”)).forEach(e ->      e.setGocDetails(“GOC0”));
     System.out.println(“modified list:”+list);

    //Converting list into map based on key here we are taking run id as key
    Map> map = list.stream().collect(Collectors.groupingBy(CheckDomain::getRunId));
    System.out.println(“map:”+map);

  //Above operations has been performed using stream same operation can be performed using parallel stream and we can take benifits of available cores
  Map> parallelMap =       list.parallelStream().collect(Collectors.groupingBy(CheckDomain::getRunId));
  System.out.println(“parallelMap:”+parallelMap);

    List gocsUsingPStream =   list.parallelStream().map(CheckDomain::getGocDetails).collect(Collectors.toList());
    System.out.println(“gocsUsingPStream:”+gocsUsingPStream);
  }

}

class CheckDomain{
      private Long runId;
      private String gocDetails;
      private Date createdDate;

      public CheckDomain() {}

      public CheckDomain(Long runId, String gocDetails, Date createdDate) {
        super();
        this.runId = runId;
        this.gocDetails = gocDetails;
        this.createdDate = createdDate;
     }
     public Long getRunId() {
            return runId;
     }
     public void setRunId(Long runId) {
        this.runId = runId;
     }
    public String getGocDetails() {
      return gocDetails;
    }
    public void setGocDetails(String gocDetails) {
        this.gocDetails = gocDetails;
    }
    public Date getCreatedDate() {
        return createdDate;
    }
    public void setCreatedDate(Date createdDate) {
        this.createdDate = createdDate;
    }

   @Override
   public String toString() {
        return “CheckDomain [runId=” + runId + “, gocDetails=” + gocDetails + “, createdDate=” +         createdDate + “]”;
    }

}

Output:

list:[CheckDomain [runId=100, gocDetails=GOC1, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=100, gocDetails=GOC1, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=101, gocDetails=GOC2, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=101, gocDetails=GOC2, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=101, gocDetails=GOC2, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=102, gocDetails=GOC3, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=103, gocDetails=GOC4, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=104, gocDetails=GOC5, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=105, gocDetails=GOC6, createdDate=Thu Mar 04 00:00:49 IST 2021]]
filteredList:[CheckDomain [runId=100, gocDetails=GOC1, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=100, gocDetails=GOC1, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=102, gocDetails=GOC3, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=103, gocDetails=GOC4, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=104, gocDetails=GOC5, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=105, gocDetails=GOC6, createdDate=Thu Mar 04 00:00:49 IST 2021]]
gocs:[GOC1, GOC1, GOC2, GOC2, GOC2, GOC3, GOC4, GOC5, GOC6]
uniqueRunIds:[100, 101, 102, 103, 104, 105]
modified list:[CheckDomain [runId=100, gocDetails=GOC0, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=100, gocDetails=GOC0, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=101, gocDetails=GOC2, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=101, gocDetails=GOC2, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=101, gocDetails=GOC2, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=102, gocDetails=GOC3, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=103, gocDetails=GOC4, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=104, gocDetails=GOC5, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=105, gocDetails=GOC6, createdDate=Thu Mar 04 00:00:49 IST 2021]]
map:{100=[CheckDomain [runId=100, gocDetails=GOC0, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=100, gocDetails=GOC0, createdDate=Thu Mar 04 00:00:49 IST 2021]], 101=[CheckDomain [runId=101, gocDetails=GOC2, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=101, gocDetails=GOC2, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=101, gocDetails=GOC2, createdDate=Thu Mar 04 00:00:49 IST 2021]], 102=[CheckDomain [runId=102, gocDetails=GOC3, createdDate=Thu Mar 04 00:00:49 IST 2021]], 103=[CheckDomain [runId=103, gocDetails=GOC4, createdDate=Thu Mar 04 00:00:49 IST 2021]], 104=[CheckDomain [runId=104, gocDetails=GOC5, createdDate=Thu Mar 04 00:00:49 IST 2021]], 105=[CheckDomain [runId=105, gocDetails=GOC6, createdDate=Thu Mar 04 00:00:49 IST 2021]]}
parallelMap:{100=[CheckDomain [runId=100, gocDetails=GOC0, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=100, gocDetails=GOC0, createdDate=Thu Mar 04 00:00:49 IST 2021]], 101=[CheckDomain [runId=101, gocDetails=GOC2, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=101, gocDetails=GOC2, createdDate=Thu Mar 04 00:00:49 IST 2021], CheckDomain [runId=101, gocDetails=GOC2, createdDate=Thu Mar 04 00:00:49 IST 2021]], 102=[CheckDomain [runId=102, gocDetails=GOC3, createdDate=Thu Mar 04 00:00:49 IST 2021]], 103=[CheckDomain [runId=103, gocDetails=GOC4, createdDate=Thu Mar 04 00:00:49 IST 2021]], 104=[CheckDomain [runId=104, gocDetails=GOC5, createdDate=Thu Mar 04 00:00:49 IST 2021]], 105=[CheckDomain [runId=105, gocDetails=GOC6, createdDate=Thu Mar 04 00:00:49 IST 2021]]}
gocsUsingPStream:[GOC0, GOC0, GOC2, GOC2, GOC2, GOC3, GOC4, GOC5, GOC6]

Popular posts from this blog

How to change column name in Dataframe and selection of few columns in Dataframe using Pyspark with example

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Solution: from pyspark.sql.functions import col # this can be done without using window function creditCardData = spark.read.json(“card_transactions.json”) useridMaxSpendDF = df.groupby(‘user_id’).max(‘amount’) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”) cond = [creditCardData.user_id == useridMaxSpen...
Read more

What is Garbage collection in Spark and its impact and resolution

-
Image
Garbage Collection Spark runs on the Java Virtual Machine (JVM). Because Spark can store large amounts of data in memory, it has a major reliance on Java’s memory management and garbage collection (GC). Therefore, garbage collection  (GC) can be a major issue that can affect many Spark applications. Common symptoms of excessive GC in Spark are: Slowness of application Executor heartbeat timeout GC overhead limit exceeded error Spark’s memory-centric approach and data-intensive applications make it a more common issue than other Java applications. Thankfully, it’s easy to diagnose if your Spark application is suffering from a GC problem. The Spark UI marks executors in red if they have spent too much time doing GC. Spark executors are spending a significant amount of CPU cycles performing garbage collection. This can be determined by looking at the “Executors” tab in the Spark application UI. Spark will mark an executor in  red  if the executor has spent more than 10% of the time in gar...
Read more

Window function in PySpark with Joins example using 2 Dataframes (inner join)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Solution: from pyspark.sql.functions import col # this can be done without using window function creditCardData = spark.read.json(“card_transactions.json”) useridMaxSpendDF = df.groupby(‘user_id’).max(‘amount’) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”) cond = [creditCardData.user_id == useridMaxSpen...
Read more