PySpark data preparation

In previous post( PySpark getting started) I was explaining how to setup PySpark dev env and some basic operations. This post is some kind of next step: then everything is in place: we have some data and we are about to start some interesting stuff like processing/modeling, but first, we have to prepare our data for this.

Starting point of my simple program

To start, we should import some dependencies from PySpark library:

from pyspark.sql import SparkSession
import pyspark.sql.functions as sqlfun

if __name__ == "__main__":
    spark = SparkSession.builder.appName("MyDataPrepareApp").getOrCreate()
    spark.sparkContext.setLogLevel('WARN')

Test data 

Let's generate some simple dataset:

print("Initial dataset:")
df = spark.createDataFrame([
    (1, "Joe", 34, None),
    (1, "Joe", 34, None),
    (2, "Huan", 36, 18000),
    (3, "Huan", 36, 18000),
    (4, None, 38, 19000),
    (5, "Sebastyan", 35, None),
    (7, "Anna", 30, 14000),
    (8, None, 24, 11000),
    (9, "Jordan", None, 15000),
    (10, None, 28, 13000),

    (1001, "ExtraLow", 1, 1),
    (1002, "ExtraBig", 80, 50000),
], schema=["id", "name", "age", "salary"])
df.show()

Output: 

Initial dataset:
+----+---------+----+------+
|  id|     name| age|salary|
+----+---------+----+------+
|   1|      Joe|  34|  null|
|   1|      Joe|  34|  null|
|   2|     Huan|  36| 18000|
|   3|     Huan|  36| 18000|
|   4|     null|  38| 19000|
|   5|Sebastyan|  35|  null|
|   7|     Anna|  30| 14000|
|   8|     null|  24| 11000|
|   9|   Jordan|null| 15000|
|  10|     null|  28| 13000|
|1001| ExtraLow|   1|     1|
|1002| ExtraBig|  80| 50000|
+----+---------+----+------+

Duplicates 

As you can see there are some(2 records with ID=1) duplicates in our dataset. Let's try to find and remove them:

# drop duplicatesdfDistinct = df.dropDuplicates()
print("Distinct:")
dfDistinct.show()

Output: 

Distinct:
+----+---------+----+------+
|  id|     name| age|salary|
+----+---------+----+------+
|1001| ExtraLow|   1|     1|
|   9|   Jordan|null| 15000|
|   4|     null|  38| 19000|
|   1|      Joe|  34|  null|
|   3|     Huan|  36| 18000|
|   8|     null|  24| 11000|
|   2|     Huan|  36| 18000|
|   7|     Anna|  30| 14000|
|1002| ExtraBig|  80| 50000|
|  10|     null|  28| 13000|
|   5|Sebastyan|  35|  null|
+----+---------+----+------+

It's better now: we don't have duplicates with ID=1 - just one record left. But we also have 2 rows which have different IDs (2 and 3) but in fact they are the same. Let's try to find duplicates for columns other than ID:

# drop duplicates other than ID columncolsExceptID = [col for col in df.columns if col != "id"]
print("Columns except [id]:", colsExceptID, ". Making distinct by them:")
dfDistinctExId = dfDistinct.dropDuplicates(colsExceptID)
dfDistinctExId.show()

Output:
Columns except [id]: ['name', 'age', 'salary'] . Making distinct by them:
+----+---------+----+------+
|  id|     name| age|salary|
+----+---------+----+------+
|   3|     Huan|  36| 18000|
|   1|      Joe|  34|  null|
|  10|     null|  28| 13000|
|   9|   Jordan|null| 15000|
|1001| ExtraLow|   1|     1|
|   7|     Anna|  30| 14000|
|   4|     null|  38| 19000|
|   8|     null|  24| 11000|
|   5|Sebastyan|  35|  null|
|1002| ExtraBig|  80| 50000|
+----+---------+----+------+

Good: just records with ID=3 remained (with ID=2 was removed)

Some basic aggregation

If we need to calculate count of duplicates and distinct records we can use aggregation functions:

# some aggregationsprint("Some basic aggregations:")
df.agg(
    sqlfun.count("id").alias("total_count"),
    sqlfun.countDistinct("id").alias("distinct_id"),
    sqlfun.countDistinct("name").alias("distinct_name")
).show()

Output: 

Some basic aggregations:
+-----------+-----------+-------------+
|total_count|distinct_id|distinct_name|
+-----------+-----------+-------------+
|         12|         11|            7|
+-----------+-----------+-------------+

Missing values

As we can see, some values in dataset are missing. Let's calculate how much records are missed for example  in AGE column:

# dealing with missed valuesprint("Percentage of missing values for [age] column")
dfDistinctExId.agg(
    (100 - sqlfun.count("age") / sqlfun.count("*") * 100).alias("age_missing")
).show()

Output:

+-----------+
|age_missing|
+-----------+
|       10.0|
+-----------+

- 10% are missing in AGE column.


We can also use more generic way to calculate missing values in several columns in "one shot":

print("Percentage of missing values for all columns")
dfDistinctExId.agg(*[  # symbol * here instruct spark to use each element of array as separate value    (100 - sqlfun.count(col) / sqlfun.count("*") * 100).alias(col + "_missing")
    for col in dfDistinctExId.columns
]).show()

Output: 

Percentage of missing values for all columns
+----------+------------+-----------+--------------+
|id_missing|name_missing|age_missing|salary_missing|
+----------+------------+-----------+--------------+
|       0.0|        30.0|       10.0|          20.0|
+----------+------------+-----------+--------------+

Filling empty values

Now when we identified missing values, we can try to fill them. Most popular approach here is to fill them by column median (mean). Let's calculate median(mean) for every column:

# calculate meanmeans = dfDistinctExId.agg(*[
    sqlfun.mean(col).alias(col + "_mean")
    for col in dfDistinctExId.columns if col != "id" and col != "name"])
print("Calculated medians(means):")
means.show()

Output: 

Calculated medians(means):
+--------+-----------+
|age_mean|salary_mean|
+--------+-----------+
|    34.0|  17500.125|
+--------+-----------+

Actually this form is good for us, but for PySpark it's better to transform it into "dictionary" format:

# creating dictionary for means

meansDict = {}
for i in range(len(means.columns)):
    colName = means.columns[i]
    colNameForDict = colName.replace("_mean", "")
    meanValue = means.first()[i]
    meansDict[colNameForDict] = meanValue
meansDict["name"] = "Unknown"print("Dictionary which will be used to fill missing values:")
print(meansDict)

Output:

Dictionary which will be used to fill missing values:
{'age': 34.0, 'salary': 17500.125, 'name': 'Unknown'}



For "name" column we can't use median value so we will be using word "Unknown".

And finally, using dictionary from above we can fill empty values:

dfFull = dfDistinctExId.fillna(meansDict)
print("With filled missing values:")
dfFull.show()

Output:

With filled missing values:
+----+---------+---+------+
|  id|     name|age|salary|
+----+---------+---+------+
|   3|     Huan| 36| 18000|
|   1|      Joe| 34| 17500|
|  10|  Unknown| 28| 13000|
|   9|   Jordan| 34| 15000|
|1001| ExtraLow|  1|     1|
|   7|     Anna| 30| 14000|
|   4|  Unknown| 38| 19000|
|   8|  Unknown| 24| 11000|
|   5|Sebastyan| 35| 17500|
|1002| ExtraBig| 80| 50000|
+----+---------+---+------+

Outliers

Great ! We don't have any empty values! But what about the other values? Some values (with IDs 1001 and 1002) looks too big and we also have some which are too small: outliers. Let's found and remove them. We're interesting here at columns AGE and SALARY. First we have to calculate quantiles: 0.5 - mean median (mean) of dataset , 0.25 - median of subset from the beginning to median, 0,75 - median of subset from median to the end..... In code I'm calculating them all (0.0, 0.25, 0.5, 0.75, 1.0) ,  bu actually we need just 2 of them: Q1(0.25) and Q3(0.75). Later we have to calculate IQR as difference between them and using IQR we can calculate lowest and highest threshold boundaries. Values are lower than lowest threshold or higher than higher threshold should be removed.

# calculate threshold boundaries for several columns(age and salary) and filter datacols = ["age", "salary"]

dfFiltered = dfFull
for currentCol in cols:
    qs = dfFull.approxQuantile(currentCol, [0.0, 0.25, 0.5, 0.75, 1.0], 0.05)
    Q1 = qs[1]
    Q3 = qs[3]
    IRQ = Q3 - Q1
    thresholdLow = Q1 - 1.5 * IRQ
    thresholdHigh = Q3 + 1.5 * IRQ
    print("For", currentCol, "low threshold is", thresholdLow, " and high threshold is", thresholdHigh)
    dfFiltered = dfFiltered.filter(
        (sqlfun.col(currentCol) > thresholdLow) & (sqlfun.col(currentCol) < thresholdHigh))

print("Filtered by thresholds:")
dfFiltered.show()

Output: 

For age low threshold is 16.0  and high threshold is 48.0
For salary low threshold is 5500.0  and high threshold is 25500.0
Filtered by thresholds:
+---+---------+---+------+
| id|     name|age|salary|
+---+---------+---+------+
|  3|     Huan| 36| 18000|
|  1|      Joe| 34| 17500|
| 10|  Unknown| 28| 13000|
|  9|   Jordan| 34| 15000|
|  7|     Anna| 30| 14000|
|  4|  Unknown| 38| 19000|
|  8|  Unknown| 24| 11000|
|  5|Sebastyan| 35| 17500|
+---+---------+---+------+


As we wanted values where had extra big/low values were removed.

The end

We removed duplicates and outliers, filled empty values - our data are ready for future processing / modeling ! :)

Just in case full code:

from pyspark.sql import SparkSession
import pyspark.sql.functions as sqlfun

if __name__ == "__main__":
    spark = SparkSession.builder.appName("MyDataPrepareApp").getOrCreate()
    spark.sparkContext.setLogLevel('WARN')

    print("Initial dataset:")
    df = spark.createDataFrame([
        (1, "Joe", 34, None),
        (1, "Joe", 34, None),
        (2, "Huan", 36, 18000),
        (3, "Huan", 36, 18000),
        (4, None, 38, 19000),
        (5, "Sebastyan", 35, None),
        (7, "Anna", 30, 14000),
        (8, None, 24, 11000),
        (9, "Jordan", None, 15000),
        (10, None, 28, 13000),

        (1001, "ExtraLow", 1, 1),
        (1002, "ExtraBig", 80, 50000),
    ], schema=["id", "name", "age", "salary"])
    df.show()


    # drop duplicates    dfDistinct = df.dropDuplicates()
    print("Distinct:")
    dfDistinct.show()

    # drop duplicates other than ID column    colsExceptID = [col for col in df.columns if col != "id"]
    print("Columns except [id]:", colsExceptID, ". Making distinct by them:")
    dfDistinctExId = dfDistinct.dropDuplicates(colsExceptID)
    dfDistinctExId.show()

    # some aggregations    print("Some basic aggregations:")
    df.agg(
        sqlfun.count("id").alias("total_count"),
        sqlfun.countDistinct("id").alias("distinct_id"),
        sqlfun.countDistinct("name").alias("distinct_name")
    ).show()

    # add new id column    print("With new created ID column")
    dfNewId = dfDistinctExId.withColumn("new_id", sqlfun.monotonically_increasing_id())
    dfNewId.show()

    # dealing with missed values    print("Percentage of missing values for [age] column")
    dfDistinctExId.agg(
        (100 - sqlfun.count("age") / sqlfun.count("*") * 100).alias("age_missing")
    ).show()

    print("Percentage of missing values for all columns")
    dfDistinctExId.agg(*[  # symbol * here instruct spark to use each element of array as separate value        (100 - sqlfun.count(col) / sqlfun.count("*") * 100).alias(col + "_missing")
        for col in dfDistinctExId.columns
    ]).show()

    ####################################    # filling missing values with MEAN    ####################################
    # calculate mean    means = dfDistinctExId.agg(*[
        sqlfun.mean(col).alias(col + "_mean")
        for col in dfDistinctExId.columns if col != "id" and col != "name"    ])
    print("Calculated medians(means):")
    means.show()

    # creating dictionary for means    meansDict = {}
    for i in range(len(means.columns)):
        colName = means.columns[i]
        colNameForDict = colName.replace("_mean", "")
        meanValue = means.first()[i]
        meansDict[colNameForDict] = meanValue
    meansDict["name"] = "Unknown"    print("Dictionary which will be used to fill missing values:")
    print(meansDict)

    # filling empty values using means from dictionary    dfFull = dfDistinctExId.fillna(meansDict)
    print("With filled missing values:")
    dfFull.show()

    ###########    # outliers    ###########
    # calculate quartiles: Q0, Q1, Q2, Q3, Q4 for [age] column    

    qs = dfFull.approxQuantile("age", [0.0, 0.25, 0.5, 0.75, 1.0], 0.05)  

    # last parameter - "accurance" it can be 0 but it will be really expensive to calculate it    

    print("quartiles: Q0, Q1, Q2, Q3, Q4 for [age] column:", qs)

    # calculate threshold boundaries for several columns(age and salary) and filter data    

    cols = ["age", "salary"]

    dfFiltered = dfFull
    for currentCol in cols:
        qs = dfFull.approxQuantile(currentCol, [0.0, 0.25, 0.5, 0.75, 1.0], 0.05)
        Q1 = qs[1]
        Q3 = qs[3]
        IRQ = Q3 - Q1
        thresholdLow = Q1 - 1.5 * IRQ
        thresholdHigh = Q3 + 1.5 * IRQ
        print("For", currentCol, "low threshold is", thresholdLow, " and high threshold is", thresholdHigh)
        dfFiltered = dfFiltered.filter(
            (sqlfun.col(currentCol) > thresholdLow) & (sqlfun.col(currentCol) < thresholdHigh))

    print("Filtered by thresholds:")

    dfFiltered.show()

Spring Boot Data JPA (with REST)

0. Intro 

It's getting more and more easier to create rest services with spring boot : now It's possible to create them with just several annotations! Let's see :)

1. Project structure

I created simple gradle project with some tests:

2.  Dependencies (build.gradle)

We need:
- spring-data-jpa   - to use auto-generated by Spring repositories
- sparing-data-rest - to expose these repositories by REST services
- lombok - to reduce code for entities classes
- h2 - as database

plugins {
   id 'org.springframework.boot' version '2.1.3.RELEASE'   id 'java'}

apply plugin: 'io.spring.dependency-management'
group = 'com.demien'version = '0.0.1-SNAPSHOT'sourceCompatibility = '1.8'
repositories {
   mavenCentral()
}

dependencies {
   implementation 'org.springframework.boot:spring-boot-starter-data-jpa'   implementation 'org.springframework.boot:spring-boot-starter-data-rest'   runtimeOnly 'com.h2database:h2'   compileOnly('org.projectlombok:lombok')
   testImplementation 'org.springframework.boot:spring-boot-starter-test'}

3. Main runner class 

Nothing interesting is here - just spring boot runner:

package com.demien.sprdata;
import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplicationpublic class SprdataApplication {

   public static void main(String[] args) {
      final SpringApplication app = new SpringApplication(SprdataApplication.class);      app.run(args);   }

}

4. Domain entities 

I create only 2 entities: "parent" : UserGroup and "child": User.

4.1. UserGroup entity 

Thanks to lombok, it's very simple, we just have to list properties. Also we have to annotate it as "@Entity" and also I'm defining named query "UserGroup.namedQueryByName"| :

package com.demien.sprdata.domain;
import javax.persistence.Entity;import javax.persistence.GeneratedValue;import javax.persistence.GenerationType;import javax.persistence.Id;import javax.persistence.NamedQuery;
import lombok.Getter;import lombok.Setter;
@Entity@Getter@Setterpublic class UserGroup {

   @Id   @GeneratedValue(strategy = GenerationType.AUTO)
   private Long id;   private String name;   private String description;
   public UserGroup() {

   }

}

4.2. User entity

It's little bit more complicated: we have to define relationship with parent: we will be joining by field groupId. And I'm also defining named query.

package com.demien.sprdata.domain;
import javax.persistence.Entity;import javax.persistence.FetchType;import javax.persistence.GeneratedValue;import javax.persistence.GenerationType;import javax.persistence.Id;import javax.persistence.JoinColumn;import javax.persistence.ManyToOne;import javax.persistence.NamedQuery;
import lombok.Getter;import lombok.Setter;
@Getter@Setter@Entity@NamedQuery(name = "User.namedQueryByName", query = "SELECT u FROM User u WHERE u.name = :name ")
public class User {

   @Id   @GeneratedValue(strategy = GenerationType.AUTO)
   private Long id;   @ManyToOne(fetch = FetchType.LAZY)
   @JoinColumn(name = "groupId")
   private UserGroup group;   private String name;   private Integer salary;
   public User() {
   }

}

5. Repositories

The magic is happening here. We just have to define interface ..... and that's it! We don't have to create the implementation - it will be created by Spring!!!

5.1. UserGroup repository 

I made it very simple: we're just extending CrudRepository and adding several methods.
"Crud" means Create, Update and Delete - so all these methods will be available in implementation which will be generated by Spring. And also 2 more methods added: findAll and count;

package com.demien.sprdata.repository;
import org.springframework.data.repository.CrudRepository;
import com.demien.sprdata.domain.UserGroup;
public interface UserGroupRepository extends CrudRepository<UserGroup, Long> {

   Iterable<UserGroup> findAll();
   long count();
}

5.2. UserRepository 

It's more complicated. First of all, we're annotation it with @RepositoryRestResource to expose methods as rest services.  Next - I'm using PagingAndSorting repository, so paging and sorting features will be available. Also I'm adding a lot of methods which will be generated by Spring:
- we can use patter [find | count] By [fieldName].
- we can even use fields of parent entity (UserGroup which is accessed by "group" field in User entity)
- we can define our own queries

package com.demien.sprdata.repository;
import java.util.List;
import org.springframework.data.jpa.repository.Query;import org.springframework.data.repository.PagingAndSortingRepository;import org.springframework.data.repository.query.Param;import org.springframework.data.rest.core.annotation.RepositoryRestResource;
import com.demien.sprdata.domain.User;
//http://localhost:8080/users/@RepositoryRestResource(collectionResourceRel = "users", path = "users")
public interface UserRepository extends PagingAndSortingRepository<User, Long> {
   Iterable<User> findAll();
   long count();
   List<User> findByName(String name);
   Long deleteByName(String name);
   Long countByGroupName(String groupName);
   // find by parent entity : Group   List<User> findByGroupName(String name);
   // defining custom query   @Query("SELECT u FROM User u WHERE u.name LIKE CONCAT('%', :name, '%') ")
   List<User> queryByName(@Param("name") String name);
   // using named query defined in entity class   List<User> namedQueryByName(@Param("name") String name);
   @Query(value = "SELECT * FROM User WHERE name = :name ", nativeQuery = true)
   List<User> nativeQueryByName(@Param("name") String name);
}

6. Testing 

And now let's test how this magic works

6.1 UserGroupRepository -  test

Here I'm testing just CRUD operations:

package com.demien.sprdata;
import static org.junit.Assert.assertEquals;import static org.junit.Assert.assertFalse;import static org.junit.Assert.assertTrue;
import java.util.ArrayList;import java.util.List;import java.util.Optional;
import org.junit.Test;import org.junit.runner.RunWith;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.autoconfigure.orm.jpa.DataJpaTest;import org.springframework.boot.test.autoconfigure.orm.jpa.TestEntityManager;import org.springframework.test.context.junit4.SpringRunner;
import com.demien.sprdata.domain.UserGroup;import com.demien.sprdata.repository.UserGroupRepository;
@DataJpaTest@RunWith(SpringRunner.class)
public class UserGroupRepositoryTest {

   @Autowired   private UserGroupRepository groupRepository;
   @Autowired   private TestEntityManager em;
   @Test   public void findAllTest() {
      final List<UserGroup> groups = new ArrayList<>();      groupRepository.findAll().forEach(group -> groups.add(group));      assertEquals(4, groups.size());   }

   @Test   public void checkUserGroupCount() {
      assertEquals(4, groupRepository.count());
   }

   @Test   public void findOne() {
      Optional<UserGroup> opGroup = groupRepository.findById(1001L);      assertTrue(opGroup.isPresent());      assertEquals("ADM", opGroup.get().getName());
      groupRepository.deleteById(1001L);      opGroup = groupRepository.findById(1001L);      assertFalse(opGroup.isPresent());
   }

   @Test   public void createNewTest() {
      final UserGroup newGroup = new UserGroup();      newGroup.setDescription("Created");      groupRepository.save(newGroup);      assertTrue(newGroup.getId() != null);
      em.flush();      final Optional<UserGroup> loaded = groupRepository.findById(newGroup.getId());      assertTrue(loaded.isPresent());      assertTrue(loaded.get().getDescription().equals("Created"));
      groupRepository.deleteById(newGroup.getId());
   }

}

6.2. UserRepository - test

And most interesting is happening here: I'm testing paging, sorting, named queries, native queries....

package com.demien.sprdata;
import static org.junit.Assert.assertTrue;
import java.util.List;
import org.junit.Test;import org.junit.runner.RunWith;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.autoconfigure.orm.jpa.DataJpaTest;import org.springframework.boot.test.autoconfigure.orm.jpa.TestEntityManager;import org.springframework.data.domain.Page;import org.springframework.data.domain.PageRequest;import org.springframework.data.domain.Sort;import org.springframework.test.context.junit4.SpringRunner;
import com.demien.sprdata.domain.User;import com.demien.sprdata.repository.UserRepository;
@DataJpaTest@RunWith(SpringRunner.class)
public class UserRepositoryTest {

   @Autowired   UserRepository userRepository;
   @Autowired   TestEntityManager em;
   @Test   public void sortTest() {
      Sort sort = new Sort(Sort.Direction.ASC, "group_id").and(new Sort(Sort.Direction.DESC, "name"));      Iterable<User> users = userRepository.findAll(sort);      User first = users.iterator().next();      assertTrue(first.getGroup().getId() == 1001L);      assertTrue(first.getName().equals("Victor"));   }

   @Test   public void pagingTest() {
      final PageRequest pageRequest = PageRequest.of(0, 2);      final Page<User> userPage = userRepository.findAll(pageRequest);      assertTrue(userPage.getNumberOfElements() == 2);      assertTrue(userPage.getTotalPages() == 4);
   }

   @Test   public void findTest() {
      List<User> users = userRepository.findByName("Joe");      assertTrue(users.size() == 1);      assertTrue(users.get(0).getName().equals("Joe"));
      Long countByGroupName = userRepository.countByGroupName("ADM");      assertTrue(countByGroupName == 3L);
      users = userRepository.findByGroupName("ADM");      assertTrue(users.size() == 3);
      users = userRepository.queryByName("a");      assertTrue(users.size() == 5);      assertTrue(users.get(0).getName().contains("a"));
      users = userRepository.namedQueryByName("Charles");      assertTrue(users.size() == 1);      assertTrue(users.get(0).getId() == 104L);
      users = userRepository.nativeQueryByName("Mario");      assertTrue(users.size() == 1);      assertTrue(users.get(0).getId() == 105L);
   }

}

7. Rest services

And final thing: let's now run our application runner and open in browser: http://localhost:8080/users/

It should be something like :

{
  "_embedded" : {
    "users" : [ {
      "name" : "Joe",
      "salary" : 100,
      "_links" : {
        "self" : {
          "href" : "http://localhost:8080/users/101"
        },
        "user" : {
          "href" : "http://localhost:8080/users/101"
        },
        "group" : {
          "href" : "http://localhost:8080/users/101/group"
        }
      }
    }, {
      "name" : "Huan",
      "salary" : 200,
      "_links" : {
        "self" : {
          "href" : "http://localhost:8080/users/102"
        },
        "user" : {
          "href" : "http://localhost:8080/users/102"
        },
        "group" : {
          "href" : "http://localhost:8080/users/102/group"
        }
      }
    }, {
      "name" : "Michael",
      "salary" : 300,
      "_links" : {
        "self" : {
          "href" : "http://localhost:8080/users/103"
        },
        "user" : {
          "href" : "http://localhost:8080/users/103"
        },
        "group" : {
          "href" : "http://localhost:8080/users/103/group"
        }
      }
    }, {
      "name" : "Charles",
      "salary" : 100,
      "_links" : {
        "self" : {
          "href" : "http://localhost:8080/users/104"
        },
        "user" : {
          "href" : "http://localhost:8080/users/104"
        },
        "group" : {
          "href" : "http://localhost:8080/users/104/group"
        }
      }
    }, {
      "name" : "Mario",
      "salary" : 200,
      "_links" : {
        "self" : {
          "href" : "http://localhost:8080/users/105"
        },
        "user" : {
          "href" : "http://localhost:8080/users/105"
        },
        "group" : {
          "href" : "http://localhost:8080/users/105/group"
        }
      }
    }, {
      "name" : "Jan",
      "salary" : 300,
      "_links" : {
        "self" : {
          "href" : "http://localhost:8080/users/106"
        },
        "user" : {
          "href" : "http://localhost:8080/users/106"
        },
        "group" : {
          "href" : "http://localhost:8080/users/106/group"
        }
      }
    }, {
      "name" : "Victor",
      "salary" : 500,
      "_links" : {
        "self" : {
          "href" : "http://localhost:8080/users/107"
        },
        "user" : {
          "href" : "http://localhost:8080/users/107"
        },
        "group" : {
          "href" : "http://localhost:8080/users/107/group"
        }
      }
    } ]
  },
  "_links" : {
    "self" : {
      "href" : "http://localhost:8080/users{?page,size,sort}",
      "templated" : true
    },
    "profile" : {
      "href" : "http://localhost:8080/profile/users"
    },
    "search" : {
      "href" : "http://localhost:8080/users/search"
    }
  },
  "page" : {
    "size" : 20,
    "totalElements" : 7,
    "totalPages" : 1,
    "number" : 0
  }
}

- HATEOAS is in place  !

8. The end

As I mentioned at the beginning, with spring-boot stack we can create rest services with DB repositories by just adding few annotations! Full source code can be downloaded from here. 

Apache Spark - getting started: batch and stream data processing using scala

0. Intro 

In this post I'm going to explain basics of Apache Spark: RDD, SparkSQL, DataFrames, SparkStreaming. Actually Spark -  it's just a library for data processing. So it can be executed without any BigData-related stuff. You can just run code from this post without any Hadoop/HDFS.

1. What is Apache Spark

from wiki:
Apache Spark is an open-source cluster-computing framework. Originally developed at the University of California, Berkeley's AMPLab, the Spark codebase was later donated to the Apache Software Foundation, which has maintained it since. Spark provides an interface for programming entire clusters with implicit data parallelism and fault tolerance.

From official documentaion:
Apache Spark is a fast and general-purpose cluster computing system. It provides high-level APIs in Java, Scala, Python and R, and an optimized engine that supports general execution graphs. It also supports a rich set of higher-level tools including Spark SQL for SQL and structured data processing, MLlib for machine learning, GraphX for graph processing, and Spark Streaming.

As you can see, there is no mentioning of Hadoop/HDFS at all. It CAN work with Hadoop/HDFS, it CAN work with cluster resource manager like YARN. But for getting familiar,  also it can be used  for processing local files in standalone mode.

2. What is Lambda architecture

From wiki:
Lambda architecture is a data-processing architecture designed to handle massive quantities of data by taking advantage of both batch- and stream-processingmethods. This approach to architecture attempts to balance latency, throughput, and fault-tolerance by using batch processing to provide comprehensive and accurate views of batch data, while simultaneously using real-time stream processing to provide views of online data. The two view outputs may be joined before presentation. The rise of lambda architecture is correlated with the growth of big data, real-time analytics, and the drive to mitigate the latencies of map-reduce.^[1]

In shorts, lambda architecture is a combination of 2 processing types:
- slow but precese - in this post we will use Spark RDD/DataFrame for it
- fast but not precise -  in this post we will use Spark Streaming for it

3. Project setup: pom.xml

Here I'm using maven with scala. In pom.xml I'm adding dependencies for spark-core, spark-sql, spark-streaming.

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.demien</groupId>
    <artifactId>sparktest</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>


    </dependencies>


</project>

4. Project structure

As you can see, we will be using 2 processing types of batch processing: RDD and DataFrame. And one type of stream processing: DStream. Apart from that we will be using SparkUtils class for creation of common sprak entities: SparkContext, SparkSession. And a generator of text files(for streaming) TextFileCreator.

5. SparkUtils

As I mentioned before, we will be using SparkUtils class for creation of common sprak entities: SparkContext, SparkSession:

package com.demien.sparktest

import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}

object SparkUtils {

  val DEF_APP_NAME = "MySparkApp"  val DEF_MASTER = "local[2]"
  //The SparkContext object - connection to a Spark execution environment and created RDDs  def getSparkContext(appName: String, master: String): SparkContext = new SparkContext(new SparkConf().setAppName(appName).setMaster(master))

  def getSparkContext(): SparkContext = getSparkContext(DEF_APP_NAME, DEF_MASTER)

  //The SparkSession - connection to dataframes and SQLs  def getSparkSession(appName: String, master: String): SparkSession = SparkSession
    .builder()
    .appName(appName)
    .master(master)
    .getOrCreate()

  def getSparkSession(): SparkSession = getSparkSession(DEF_APP_NAME, DEF_MASTER)


}

6. Batch processing: RDD

From official documentation:

The main abstraction Spark provides is a resilient distributed dataset (RDD), which is a collection of elements partitioned across the nodes of the cluster that can be operated on in parallel. RDDs are created by starting with a file in the Hadoop file system (or any other Hadoop-supported file system), or an existing Scala collection in the driver program, and transforming it. Users may also ask Spark to persist an RDD in memory, allowing it to be reused efficiently across parallel operations. Finally, RDDs automatically recover from node failures.

RDD Example: 

package com.demien.sparktest.batch

import com.demien.sparktest.SparkUtils
import org.apache.spark.rdd.RDDwith

// https://spark.apache.org/docs/2.3.0/rdd-programming-guide.htmlobject RddExample extends App {

  val sc = SparkUtils.getSparkContext()
  val file = sc.textFile("src/main/resources/sample.txt")
  val words: RDD[String] = file.flatMap(l => l.split(" ")).filter(w => w.length > 1)
  val pairs: RDD[(String, Int)] = words.map(s => (s, 1)) // [the, of, the] => (the, 1) (of, 1) (the, 1)  val counts: RDD[(String, Int)] = pairs.reduceByKey((a, b) => a + b) // (the, 1) (of, 1) (the, 1) => (the, 2) (of, 1)  val countByWord: RDD[(Int, String)] = counts.map(p => p.swap) // (the, 2) (of, 1) => (2, the) (1, of)  val countByWordSorted: RDD[(Int, String)] = countByWord.sortByKey(false)
  val top5 = countByWordSorted.take(5)

  top5.foreach(p => println(p))
}

- I added comments and  datatypes for RDD variables(which of cource are not needed here) to make it more clear. This RDD example is processing sample text file - it's just a text from ApacheSpark wiki. We are splitting text into words, creating for every work "paired object" with word itself and number 1. After that, we are groupping these pairs using word as a key and counting provided numbers.

Results:
(55,the)
(46,of)
(43,Spark)
(39,and)
(24,in)

- as you can see, most popular word(excluding "the", "of", "and", "in") in Spark wiki is "Spark" :)

7. Batch processing: DataFrame/Spark SQL

From official documentation:

A DataFrame is a Dataset organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R/Python, but with richer optimizations under the hood. DataFrames can be constructed from a wide array of sources such as: structured data files, tables in Hive, external databases, or existing RDDs. The DataFrame API is available in Scala, Java, Python, and R. In Scala and Java, a DataFrame is represented by a Dataset of Rows. In the Scala API, DataFrame is simply a type alias of Dataset[Row]. While, in Java API, users need to use Dataset<Row> to represent a DataFrame.

DataFrames - are the structured datasets, so as a sample file we will be using not TEXT file but  JSON like this:

{
  "name": "Keeley Bosco",
  "email": "katlyn@jenkinsmaggio.net",
  "city": "Lake Gladysberg",
  "mac": "08:fd:0b:cd:77:f7",
  "timestamp": "2015-04-25 13:57:36 +0700",
  "creditcard": "1228-1221-1221-1431"}
{
  "name": "Rubye Jerde",
  "email": "juvenal@johnston.name",
  "city": null,
  "mac": "90:4d:fa:42:63:a2",
  "timestamp": "2015-04-25 09:02:04 +0700",
  "creditcard": "1228-1221-1221-1431"}

DataFrame example:

package com.demien.sparktest.batch

// https://spark.apache.org/docs/latest/sql-programming-guide.html
import org.apache.spark.sql.SparkSession

object DataFrameExample extends App {

  val spark = SparkSession
    .builder()
    .appName("Spark SQL basic example")
    .config("spark.master", "local")
    .getOrCreate()

  // For implicit conversions like converting RDDs to DataFrames  import spark.implicits._


  val df = spark.read.json("src/main/resources/people.json")
  df.printSchema()
  df.createOrReplaceTempView("people")
  val sqlDF = spark.sql("SELECT * FROM people where email like '%net%' ")
  sqlDF.show()

  case class Person(name: String, email: String, city: String, mac: String, timestamp: String, creditcard: String)

  val peopleDS = spark.read.json("src/main/resources/people.json").as[Person]
  val filteredDS = peopleDS.filter(p => p.email != null && p.email.contains("net"))
  filteredDS.show()


}

We are using SparkSQL to query our structured dataset(DataFrame) for people which have "%net%" in their emails. Also, at the end we are doing the same thing again, but using using DataSet api.
Or cource, in both cases results are the same:

+---------------+-------------------+--------------------+-----------------+----------------+--------------------+
|           city|         creditcard|               email|              mac|            name|           timestamp|
+---------------+-------------------+--------------------+-----------------+----------------+--------------------+
|Lake Gladysberg|1228-1221-1221-1431|katlyn@jenkinsmag...|08:fd:0b:cd:77:f7|    Keeley Bosco|2015-04-25 13:57:...|
|           null|1228-1221-1221-1431|emery_kunze@rogah...|3a:af:c9:0b:5c:08|Celine Ankunding|2015-04-25 14:22:...|
+---------------+-------------------+--------------------+-----------------+----------------+--------------------+


Unfortunatelly, spark is not showing full values, but these email values  are:

"katlyn@jenkinsmaggio.net, "emery_kunze@rogahn.net"

- emails which contain "net".

8. Stream processing: DStream

From official documentation:

Spark Streaming is an extension of the core Spark API that enables scalable, high-throughput, fault-tolerant stream processing of live data streams. Data can be ingested from many sources like Kafka, Flume, Kinesis, or TCP sockets, and can be processed using complex algorithms expressed with high-level functions like map, reduce, join and window. Finally, processed data can be pushed out to filesystems, databases, and live dashboards. In fact, you can apply Spark’s machine learning and graph processing algorithms on data streams.

8.1 Stream processing: DStream: TextFileCreator

To simulate stream of data, we will create the simple application which is creating text file every 10 seconds. As a source for this file I will be using again text from ApachSpark wiki. 

package com.demien.sparktest

import java.io.FileWriter
import java.util.Date

import scala.io.Source
import scala.util.Random

object TextFileCreator extends App {

  val listOfLines = Source.fromFile("src/main/resources/sample.txt").getLines.toList
  val rnd = new Random()

  while (true) {

    val fileName = new Date().getTime
    val fullFileName = "data/" + fileName + ".txt"    val fw = new FileWriter(fullFileName, true)
    println("writing to " + fullFileName)

    val linesCount = rnd.nextInt(20) + 5    for (i <- 1 to linesCount) fw.write(listOfLines(rnd.nextInt(100)) + "\n")

    fw.close()
    Thread.sleep(10000)

  }

}

8.2 Stream processing: DStream: Streaming example itself

Our application will be monitoring "data" folder for new files. When new file is received - it will be processed. To simulate some statefull activity we will be using function:  specFunc - the point is to constantly calculate count of words. 

package com.demien.sparktest.stream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, State, StateSpec, StreamingContext}

object DStreamExample extends App {


  val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
  val ssc = new StreamingContext(conf, Seconds(10))
  ssc.checkpoint("spark-checkpoint")

  val lines = ssc.textFileStream("data")
  val words = lines.flatMap(_.split(" "))
  val pairs = words.map(word => (word, 1))
  val wordCounts = pairs.reduceByKey(_ + _)


  def specFunc = (key: String, value: Option[Int], state: State[Int]) => {
    var newState = state.getOption().getOrElse(0)
    var newValue = value.getOrElse(1)
    newState = newState + newValue
    state.update(newState)
    (key, newValue)
  }

  val spec = StateSpec.function(specFunc).timeout(Seconds(30))

  val wordsMapped = wordCounts.mapWithState(spec)

  // top 10  wordsMapped.stateSnapshots().foreachRDD(rdd => {
    rdd.map(e => (e._2, e._1)).sortByKey(false).take(10).foreach(e => print(e._1, e._2))

  })

  ssc.start() // Start the computation  ssc.awaitTermination() // Wait for the computation to terminate
}

8.3. Stream processing: DStream: Execution

Of course, we have to run both: TextFileCreator and DStreamExample.

TextFileCreator is creating files:
writing to data/1524921644177.txt
writing to data/1524921654262.txt
writing to data/1524921664264.txt
writing to data/1524921674266.txt
writing to data/1524921684267.txt
writing to data/1524921694268.txt
writing to data/1524921704270.txt


And DStreamExample is processing them and counting words:

.....
(17,a)(16,Spark)(15,)(14,the)(10,Apache)(10,of)(7,//)(5,is)(5,can)(5,in)
(17,)(17,a)(17,Spark)(14,the)(10,Apache)(10,of)(7,//)(5,is)(5,can)(5,in)
(25,)(24,the)(21,Spark)(21,a)(19,of)(12,Apache)(10,can)(10,as)(10,and)(9,is)
(35,)(27,the)(25,Spark)(25,a)(20,of)(17,and)(14,Apache)(12,as)(12,//)(10,can)
(45,)(27,the)(26,Spark)(25,a)(20,of)(17,and)(14,Apache)(12,as)(12,//)(10,can)

As you can see, results are similar to what we had in RddExample: most popular words are Spark and Apache.

9. The end. 

As you can see, to try Apache Spark you don't need Hadoop/Yarn - it's possible to run it in a standalone mode without all these compicated things. Source code can be downloaded from here.