Saturday, 4 December 2021

This week 3/2021 - springSecurity Rest basic controller

In this short post I'd like to present a simple configuration of Spring Boot application serving stateless service using basic authentication.

Below a web security configurer implementation including all possible ways to define annotation rule matchers (pre, post processing and jsr250 specification)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@EnableGlobalMethodSecurity(
        prePostEnabled = true,
        securedEnabled = true,
        jsr250Enabled = true)
public class SecurityConfig extends WebSecurityConfigurerAdapter {

    @Override
    protected void configure(final HttpSecurity http) {
        sessionSettings()
                .andThen(this::headersSecurity)
                .andThen(this::accessRules)
                .unchecked()
                .accept(http);
    }

    private void headersSecurity(final HttpSecurity http) throws Exception {
        http.sessionManagement()
            .sessionCreationPolicy(SessionCreationPolicy.STATELESS)
            .and()
            .csrf()
            .disable()
            .httpBasic()
            .realmName("App");
    }


    private CheckedConsumer<HttpSecurity> sessionSettings() {
        return http -> http
                .sessionManagement()
                .sessionCreationPolicy(STATELESS);
    }

    @Override
    protected void configure(final AuthenticationManagerBuilder auth) throws Exception {
        auth.inMemoryAuthentication()
                .withUser("client")
                .password(passwordEncoder().encode("admin"))
                .roles("CLIENT");
    }
...
}

What is important in code is to define session creation policy.

Then it is possible to implement standard resource.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
@RestController
@RequestMapping("/service")
public class Endpoint {

    @PostMapping("/endpoint")
    @Secured("ROLE_CLIENT")
    public String endpoint(@RequestBody final String req) {
        return "test";
    }
}

Resources:

[1] - Spring Security

Monday, 30 August 2021

This week 2/2021 - Testing Spark Application

Last year started my adventure with Big Data and Hadoop ecosystem. I created my first simple Java application for Spark. It worked but I felt some lack. Where are tests for that? I looked around and I couldn't find any good example of module testing. I was looking for test example for hexagon application, this time in Spark, so: mocks on input and output and call facade main method to check behaviour.

Finally I found a solution. Like standard business application I extracted code which is a point of contact with external world and mocked it. In my case I was testing a simple application which consumes message from Kafka, transforms it and stores a result in one of two tables.
Bellow I presented a simple class dependence which correspond to my application.

The KafkaStream and HadoopRepository are input and output of my application. On the top of them there are services and facade where are implemented transformation and DAG model creation.
 
To increase difficulty of this task I created my first test in scala.
 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class TransformationSpec extends AnyFlatSpec 
                            with BeforeAndAfter 
                            with BeforeAndAfterEach
                            with should.Matchers 
                            with GivenWhenThen {

  private val master = "local[3]"
  private val appName = "test-app"
  private val batchDuration = Seconds(1)

  private val kafkaStream = mock[KafkaStream]
  private val rowRecordRepositoryMock = mock[HadoopRepository[RowRecord]]
  private val errorRecordRepositoryMock = mock[HadoopRepository[ErrorRecord]]
  private val serviceA: ServiceA = new ServiceA(rowRecordRepositoryMock, errorRecordRepositoryMock)
  private val serviceB: ServiceB = new ServiceB(kafkaStream)
  private val underTest: TransferDataService = new TransferDataService(serviceA, serviceB)

  private var sc: SparkContext = _
  private var ssc: StreamingContext = _
  private val lines = mutable.Queue[RDD[ConsumerRecord[String, ReceivedMsg]]]()

  before {
    Mockito.reset(kafkaStream, rowRecordRepositoryMock, errorRecordRepositoryMock)
    val conf = new SparkConf()
      .setMaster(master)
      .setAppName(appName)
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    ssc = new StreamingContext(conf, batchDuration)
    sc = ssc.sparkContext
    mockKafkaStream
  }

  after {
    if (ssc != null) {
      ssc.stop()
    }
  }
...
 
In example I am building application (lines 11-18) - injecting dependences. Then I define class variables for Spark context and Spark Streaming context and finally a queue to simulate message receiving.
 
In "before" method for each test state of all mocks are reset, Spark configuration, context and mock KafkaStream with specific behaviour are created. Below is shown how messages are added to queue and how KafkaStream is mocked using MockitoSugar library.
 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
  private def receivedMessage(msg: ReceivedMsg) = {
    lines += sc.makeRDD(Seq(new ConsumerRecord("test", 0, 0, "", msg)))
  }

  private def mockKafkaStream = {
    MockitoSugar.doAnswer(() => {
      new JavaInputDStream[ConsumerRecord[String, ReceivedMsg]](
        ssc.queueStream(lines, true)
      )
    }).when(kafkaStream).createDirectStream(any());
  }
 
Now it is all ready to write first test case. I chose FlatSpec style. In "given" section I refilled message queue. In "when" section I called facade to initiate DAG model. I started Spark Streaming context and I run process.
In "then" section I asserted mock and captors. Application uses many threads so I couldn't call assertions just like that. To solve this problem scalatest provides method "eventually" to check conditions with some interval until match condition or meet timeout.
 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
"Deserialized message" should "be stored in target table" in {

  Given("deserialized message")
  receivedMessage(new ReceivedMsg())

  When("Processing")
  underTest.process(new JavaStreamingContext(ssc));
  ssc.start()

  Then("message is stored by n records")
  eventually(timeout(2 second)) {
    val captor: ArgumentCaptor[JavaRDD[RowRecord]] = ArgumentCaptor.forClass(classOf[JavaRDD[RowRecord]])
    verify(rowRecordRepositoryMock)
      .save(captor.capture())

    val records = captor.getValue.collect()
    records should have size (4)

    verify(errorRecordRepositoryMock, never())
      .save(any())

    verify(kafkaStream)
      .commit(any(), any());
  }
}
 
For me as Java Developer it new experience to write test in scala and for Spark. It wasn't big challenge when I used first 3 sources.
Concurrently I am writing some simple scripts in python. This is new area for me so the next step will be testing Python scripts for Spark.

To prepare this example I used scala 2.12, Spark 3.0.2, Mockito-scala 1.12.6  and scalatest 3.2.7

Sources:
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Saturday, 28 August 2021

This week 1/2021 - RxJava 3.x - back pressure

I got deep dive into RxJava tutorial and train a lot of possible cases. However most interesting for me is back pressure feature. 

Going throw the mechanic of RxJava for this case, at the moment of subscription generator is initialized (line 3), then subscriber onStart method is called (line 13) and first elements are generated (depending of buffer size).

In onStart method should be "request" called defining how many elements should be generated a the begging. If "request" method is missing this result no further action.
In the following example there is no difference if we ask for more elements than one, generator always waits to buffer be empty in 75% (the level when buffer is refilled is defined in BaseObserveOnSubscriber) and then start to generate.
When calling onComplete and onError flow is intermittent.


 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
public void backpressure() {
    final Flowable<Integer> generate = Flowable.generate(
            () -> "INITIALIZED", //initial state
            (state, emitter) -> { //current state
                // do some producing task
                emitter.onNext(product.get());
                return "RUNNING"; // next state
            }
    );
    
    final DefaultSubscriber<Integer> subscriber = new DefaultSubscriber<Integer>() {
        @Override
        protected void onStart() {
            request(1);
        }

        @Override
        public void onNext(final Integer o) {

            request(1);
            // do some consumer work
  
        }

        @Override
        public void onError(final Throwable throwable) {
            
        }

        @Override
        public void onComplete() {
        }
    };
    int bufferSize = 6;
    generate.observeOn(Schedulers.newThread(), false, bufferSize)
            .subscribe(subscriber);
}
 
I was training on version 3.0.11 of RxJava.

Sources:



Sunday, 27 December 2020

This week 7/2020 - Julia

This post describing my first feel when I completed a Julia basic course.

I am experienced Java developer but I have also osculation with C/C++, Python and Octave languages. For me Julia has something from all those languages.

Linear Algebra support:

Octave:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
A = eye(1,2)
%Diagonal Matrix
%
%   1   0

B = eye(3,2)
%Diagonal Matrix
%
%   1   0
%   0   1
%   0   0
C = [A
B]
%C =
%
%   1   0
%   1   0
%   0   1
%   0   0

Julia:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
using LinearAlgebra

A = 1*  Matrix(I,1,2)
# 1×2 Array{Int64,2}:
#  1  0

B = 1*  Matrix(I,3,2)
# 3×2 Array{Int64,2}:
#  1  0
#  0  1
#  0  0

C = [A
       B]
# 4×2 Array{Int64,2}:
#  1  0
#  1  0
#  0  1
#  0  0

There is also similarity when multiply matrixes. Julia support ex. operations f(A) = A*A and f.(A) whene every A[i,j] * A[i,j].

Syntax similarity to Python:

Julia:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# for loop
for i in 1:10, j in 1:20
    println("Hi $i , $j")
end
for item in items
    println("Hi $item")
end

# function definition
function power(x)
# last element is returnes - the same as in Python
    x^2
end

# other options to define function 
power(x) = x^2

power = x -> x^2

# immutable sorting
sort(x)

#mutable sorting
sort!(x)

Overload operators:

Python:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
class String:
    def __init__(self, x=""):
        self.x = x

    def __add__(self, other):
        return self.x == other.x


p1 = String("test1")
p2 = String("test1")

print(p1 + p2)

Julia:

1
2
3
4
5
6
import Base: +

+(x::String, y:: String) = x == y

# returns boolean value
x+y 

Performance:

Benchmarks which I saw in course [3] shows that Julia has similar or a little better performance than C code and this about 2 orders of magnitude than Python.


Resources:

[1] https://julialang.org/

[2] Introduction to Julia (for programmers)

[3] Parallel Computing


Saturday, 12 September 2020

This week 6/2020 - Neo4j

In this article in a nutshell I am describing the Neo4j. This article include topics:
  • a short description of database,
  • in what cases it is worth to consider use of graph database,
  • what are advantages comparing to relational database,
  • a short description of "graph SQL" - cypher,
  • a few examples of queries in cypher,
  • a shortcut how to run Java project with Spring Boot and Spring Data dependences.
 
Neo4j is a graph database. It is transactional and ACID compliant with native graph storage and processing. It use graph SQL language called Cypher dedicated for graph databases.

Graph databases can be used everywhere where there is a need to archive a graph dependency between objects, so I could say in most cases I know. 
Comparing to relational databases, native storing and processing have advantage that matching queries are executed faster than relational queries with exponential cost.
 
Taking a simple case of customers using some services there is a relation many to many.




In relational database it is required to have a matching table where there are ids of services and using them customers. To connect all customers with single service it is required to find service id then in matching table find customer ids and then in third one find customers.
In graph database every service Node (every object is a node - equivalent of table) stores direct Relation to Node customer. This requires using more storage but it is much faster than matching table relations. Other advantages are:
- auto extending schema model as in other NoSQL databases - adding data of node, relation or property in node or relation schema is automagically extended,
- handle "graph SQL" called Cypher.
 
Cypher is a dedicated language for graph database. Below I have placed a few examples.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
-- simple select from table
MATCH  (c:Human) 
WHERE id = 1 
RETURN c;
-- equivalent in SQL
SELECT * 
FROM Human 
WHERE id = 1;

-- simple relation
MATCH (p:Person) - [r:ACTED_IN] -> (m:Movie)
WHERE  p.name = 'Tom' 
RETURN p, r, m;
-- equivalent in SQL
SELECT * 
FROM Person p
JOIN Relation r on p.id = r.person_id 
JOIN Movie m on m.id = r.movie_id
WHERE p.name = 'Tom'
AND r.type = 'ACTED_IN'

In more complicated case when there is a need to create chain of relations, ex. who is above employee. Is it typical graph case? In Oracle PL/SQL there is something called "CONNECT BY" query construction but how is in other databases, truly I don't know. In MySQL I saw a recurrent procedure storing each level in temporary table, so how is in Cypher?

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
-- this returns supervisors and their supervisor
MATCH 
path = (n:Person)-[r:REPORTS_TO*]->(s) 
WHERE n.name = 'Tom' 
RETURN s 
ORDER BY length(path)

-- case with subordinates by supervisor id
MATCH 
path = (n)-[r:REPORTS_TO*]->(s:Person) 
WHERE id(s) = 12 
RETURN n 
ORDER BY length(path)

and a few other useful queries

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
-- delete all schema
MATCH (n) DETACH 
DELETE n

-- create modes and data set
CREATE (:Car:Vehicle { type: "Van"}) <-[:DRIVES ] - (:Human {name: "Basia"})


-- update data
MATCH (h:Human) - [d:DRIVES] -> (c:Car:Vehicle)
WHERE
  c.type = "Van"
SET
  h.name = "Ula",
  c.productionYear = "1982"
RETURN h,d,c

-- constraint - unique field
CREATE CONSTRAINT ON (h:Human)
ASSERT h.name IS UNIQUE

-- delete data matching query
MATCH  (c:Car:Vehicle) <-[d:DRIVES ] - (h:Human)
DELETE c,d , h

 

Spring Boot  project.

In Spring Boot with Spring Data it is required only to add spring-boot-starter-data-neo4j artefact, neo4j properties in path spring.data.neo4j.*, @EnableNeo4jRepositories in configuration and it is possible to create node entities. 

In background there is added dependency to org.neo4j:neo4j-orgm-* artefacts and spring-data-neo4j and also org.neo4j.driver artefact.

I was working on:
- docker image of neo4j v.: 4.1.1 without auth.
- JDK11
- Spring Boot v.:2.2.4

Added neo4j dependences was:
- org.neo4j.driver v.: 4.0.0
- org.neo4j:neo4j-ogm-* v.:3.2.6
- spring-data-neo4j v.: 5.2.4-RELEASE


Saturday, 29 August 2020

This week 5/2020 - Docker tools

This article is a shortcut of docker tools. These tools are commonly used in micro-service architecture:

  • Docker
  • Docker Compose

Docker

is a platform to run application using containers. Containers are created on basis of images created incrementally similar to code repositories, layer after layer.
Container is a environment to run isolated application. It doesn't use their own operating system as virtualized machines. Container share it with host that's why container stand up in a seconds and is lighter for physical machine instead of stand up minutes as virtual machine. That's why docker is commonly used to create instances of application.

Docker can be used interactively, from console. Below most useful commands:

docker ps - show all running container
docker images - show images in local repository
docker run -d [image_name] - run image in daemon mode
docker exec -it [container_id] "[command to run in the container, ex /bin/sh]" - plug in and execute command on specific container
docker container logs [container_id] - print logs from container
docker pull [image name] - pull image from external images repository

but the biggest benefit of docker is that can be used by scripts, so all process is repeatable and can be automatized. Default docker file is Dockerfile. Below some example:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# base image to this build
FROM openjdk:8-jdk-alpine

# define what directory should be mounted to host 
# - mounted directories are created in /var/lib/docker/volumes
VOLUME /tmp

# only inform what ports can expose application
EXPOSE 8080

# define variable
ARG JAR_FILE=target/*.jar


# define variable using environment variable, or "v.1.0.0" if not defined. 
#ENV override ARG variable. Example execution with variable: 
# $ docker build --build-arg CONT_IMG_VER=v2.0.1 .
ENV SOME_ENV_VAR ${CONT_IMG_VER:-v1.0.0}


#copy file from host to container storage
COPY ${JAR_FILE} app.jar

#copy file from host to container storage, but comparing to COPY 
# can also get file from url and extract tar file
ADD ${JAR_FILE} app.jar

# run command in container
RUN uname -a

# health check command - docker is checking if application is working properly
HEALTHCHECK --interval=5m --timeout=3s --retries=5 \
  CMD curl -f http://localhost/ || exit 1


# run application as goal of this image
ENTRYPOINT ["java","-Djava.security.egd=file:/dev/./urandom","-jar","/app.jar"]


having Dockerfile it is executed command:

docker build

and then if this image have to be pushed to remote repository

docker image tag [image_tag_name] [docker repository path]
docker push [docker repository path]

There is also possible to start a repository on docker container, executing a command:

docker run -d -p 5000:5000 --name registry registry:2

and to remove container registry container

docker container stop registry && docker container rm -v registry


Docker Compose

it is a tool to stand up a few containers on basic of docker-compose.yml file. Tool manages with dependences between containers, so by one command it is possible to run many services (containers). Below a few most useful commands:

docker-compose build - build images included in file docker-compose.yml
docker-compose up -d - run containers in daemon mode
docker-compose down - stop containers
docker-compose logs - print logs from containers

and docker-compose.yml file example:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# version of file format
version: "3.3"
# definition of services (container templates)
services:
 #name of service 
  mongoDB:
# image name - this image is retrieved from remote repository  
    image: library/mongo:4.4.0
# container name     
    container_name: "mongoDBcontainerName"
# what if application is dead    
    restart: on-failure
# ports which should be exposed to host (host port: container port)    
    ports:
    - 27017:27017
# images have defined variables, this way are defined their values
    environment:
      MONGO_INITDB_ROOT_USERNAME: sboot
      MONGO_INITDB_ROOT_PASSWORD: example
      MONGO_INITDB_DATABASE: test
# storage mapping ( host : containers path : access mode)      
    volumes:
      - ./src/main/sql/mongo-init.js:/docker-entrypoint-initdb.d/mongo-init.js:ro

  app:
# build properties - this service will be built 
    build:
# where is context path on host    
      context: ./
# docker file      
      dockerfile: Dockerfile
    container_name: "myApp"      
# definition of depending on services      
    depends_on:
      - mongoDB
# this defines in container dns names for depending on services
    links:
      - mongoDB

To prepare this article I used:

  • Docker in version 19.03.6 - provided by system
  • Docker Compose in version 1.17.1 - provided by system

Resources:
[1] Docker

Thursday, 27 August 2020

This week 4/2020 - Machine Learning - part II

This article is a continuation of Machine Learning series. I am presenting a few advices presented by Andrew Ng on coursera course. They are useful when building Machine Learning System (MLS). What is about this article:
  • how prepare data,
  • how to debug it,
  • what are skewed classes,
  • how to carry out ceiling analysis.
 
Preparing data set:
On small set off data (up to ~ 10-100 000 records) it is recommended to split randomized data set in following proportions:
  • 60% -  training records - used to train algorithm to find θ factors giving lowest cost.
  • 20% - cross validation records - to select best configuration of algorithm, ex. for Neutral Network (NN) to check how many layers should have network or to reduce useless features.
  • 20% - test records - to define performance of MLS.
In case of big volume of data set (above 100 000 records) it is recommended to change proportions, to respectively 92% /4%/4%.

Debugging MLS:
To improve MLS it is good to perform error analysis that's why consider:
- usage of more training examples,
- change set of features (less/more/different),
- adding polynomial features,
- change lambda value in regularization factor,
- change number of nodes or layers (refers to NN).

Size of training set -  below I added chart showing dependency between cost function and used records in training set (learning curve).


 
On the left chart it can be noticed that for high bias when added more data not decrease high error. However when function is complicated it can be observed a huge error gap but when it is added more data it slowly decrease for cross-validation data. 
This can be manipulated by changing a set of features (less/more). Below I added chart about dependency between cost (error) and complexity of wanted function and examples of function for one set of data.
 
 
 
 
How exactly this is done? At first function is trained for training data and then 
cross-validation data error is calculated for a few configurations of features.
When it is observed high bias it can mean that wanted function is too simple to prepared data set. It can be required to add new features or create polynomial features from existing features.
When it is observed high variance it can mean that wanted function is too complex. It can be required to remove some features.
 
It is possible to manipulate bias and variance by changing λ of regularization factor. Below I added 3 charts. For very big λ, just right and λ close or equal 0.


It is noticed that too big λ create almost constant function. When λ is close to 0, regularization factor is negligibly small and can be skipped.
 

Skewed classes
This term refers to situation when set of data of one category is much larger then set of other category, ex. for binary output, if there is 99 % of examples for "true" category and 1% of examples for "false" category. Then creating logistic regression algorithm and other system returning always "true" it is no so big difference between them. At least 1% of difference in effectiveness - not so bad but systems significantly different.
That's why to compare systems like this they are defined terms: 
- true positive, 
- true negative, 
- false negative,
- false positive
described on draft below:


 
and measures:
- precision - calculate ratio between true positive and false positive
$$ precision = \frac{TP}{TP + FP} $$

- recall - calculate ratio between true positive and false negative

$$recall = \frac{TP}{TP+FN} $$

What gives a measure for factors precision (P) and recall (R)
$$ F_1score = 2* \frac{P*R}{P+R} $$
so  bigger score means better system.
 
The last term in this article is ceiling analysis - this is more economic term, because focuses on whole system as a set of MLS modules working in pipeline.
This analysis answers for question which module should be improved to get higher accuracy of the application.