scala play json traversal

Another night on the startup, I learned something about json traversal using play json library.

Play Json api is cool, for example I can convert the json string to Json using Json.parse("json string").

The fun part is when I had to traverse the Json object. For example I want to make sure that the json response I received from my HTTP server has the proper values.

So, here is the example of traversing the given list of json objects,

scala> import play.api.libs.functional.syntax._
import play.api.libs.functional.syntax._

scala> import play.api.libs.json._
import play.api.libs.json._

scala> implicit val productReader = (
     |         (__ \ "id").read[Long] and
     |       (__ \ "name").read[String] and
     |         (__ \ "brand").read[String] and
     |         (__ \ "price").read[Double] and
     |         (__ \ "category").read[String]and
     |         (__ \ "marketplace").read[String] and
     |         (__ \ "date").read[Long]
     |       ) tupled;
warning: there were 1 feature warning(s); re-run with -feature for details
productReader: play.api.libs.json.Reads[(Long, String, String, Double, String, String, Long)] = play.api.libs.json.Reads$$anon$8@711cbc4c

scala> val x = """[{"0":{"price":100.0,"releasedOn":1419783511,"brand":"Palace Skate Boards","category":"Men","name":"Tops","marketplaceName":"Oodni boutique","id":1}}, {"1":{"price":200.0,"releasedOn":1419783519,"brand":"Steven Wilson","category":"Metal","name":"Hands.Can not. Erase","marketplaceName":"Kscope","id":2}}, {"2":{"price":200.0,"releasedOn":1419783519,"brand":"sleepmakeswaves","category":"Rock","name":"And So We Destroyed Everything","marketplaceName":"Bird's Robe Records","id":3}}]"""
x: String = [{"0":{"price":100.0,"releasedOn":1419783511,"brand":"Palace Skate Boards","category":"Men","name":"Tops","marketplaceName":"Oodni boutique","id":1}}, {"1":{"price":200.0,"releasedOn":1419783519,"brand":"Steven Wilson","category":"Metal","name":"Hands.Can not. Erase","marketplaceName":"Kscope","id":2}}, {"2":{"price":200.0,"releasedOn":1419783519,"brand":"sleepmakeswaves","category":"Rock","name":"And So We Destroyed Everything","marketplaceName":"Bird's Robe Records","id":3}}]

scala> val jsonArray = Json.parse(x).as[JsArray]
jsonArray: play.api.libs.json.JsArray = [{"0":{"price":100.0,"releasedOn":1419783511,"brand":"Palace Skate Boards","category":"Men","name":"Tops","marketplaceName":"Oodni boutique","id":1}},{"1":{"price":200.0,"releasedOn":1419783519,"brand":"Steven Wilson","category":"Metal","name":"Hands.Can not. Erase","marketplaceName":"Kscope","id":2}},{"2":{"price":200.0,"releasedOn":1419783519,"brand":"sleepmakeswaves","category":"Rock","name":"And So We Destroyed Everything","marketplaceName":"Bird's Robe Records","id":3}}]

scala> jsonArray \\ "0"
res14: Seq[play.api.libs.json.JsValue] = ListBuffer({"price":100.0,"releasedOn":1419783511,"brand":"Palace Skate Boards","category":"Men","name":"Tops","marketplaceName":"Oodni boutique","id":1})

scala> jsonArray(0)
res3: play.api.libs.json.JsValue = {"0":{"price":100.0,"releasedOn":1419783511,"brand":"Palace Skate Boards","category":"Men","name":"Tops","marketplaceName":"Oodni boutique","id":1}}

scala> jsonArray(0).\("0")
res4: play.api.libs.json.JsValue = {"price":100.0,"releasedOn":1419783511,"brand":"Palace Skate Boards","category":"Men","name":"Tops","marketplaceName":"Oodni boutique","id":1}

scala> jsonArray(0).\\("0")
res5: Seq[play.api.libs.json.JsValue] = List({"price":100.0,"releasedOn":1419783511,"brand":"Palace Skate Boards","category":"Men","name":"Tops","marketplaceName":"Oodni boutique","id":1})

scala> jsonArray(0).\("0").\("price")
res11: play.api.libs.json.JsValue = 100.0

scala> jsonArray(0).\("0").\("name")
res12: play.api.libs.json.JsValue = "Tops"

scala> jsonArray(0).\("0").\("brand")
res13: play.api.libs.json.JsValue = "Palace Skate Boards"

scala> ((jsonArray \\ "0").map(on => on \ "name").map(_.as[String])).asInstanceOf[Seq[String]](0)
res33: String = Tops

Reference

https://www.playframework.com/documentation/2.5.x/ScalaJson

consuming scala play Enumerator/Stream

I was working on a startup using scala play app and was unit testing the response from Play Middlerware which is basically returningFuture[SimpleResult] with body as Enumerator[Array[Byte]].

If the response has json a body, Enumerator won't give me the body as flat string, rather a list of chunks. So, Enumerator is a simply a async collection aka Stream which needs to be consumed by someone else( Iteratee).

Stream source

For example, song is a collection of buffered chunks of audio clips.

SBT_SCALA_VERSION=2.10.4 play console

scala> import play.api.libs.iteratee.{Enumerator, Iteratee}
import play.api.libs.iteratee.{Enumerator, Iteratee}

scala> val songStream: Enumerator[String] = Enumerator("first 5 minutes ", "second 5 minutes ", "last 5 minutes")
songStream: play.a.l.i.Enumerator[String] = play.api.libs.i.Enumerator$$anon$19@24142c37

 // Enumerator can also be created using the stream
scala> val songStream: Enumerator[String] = Concurrent.unicast[String](onStart = stream => {
  stream.push("first 5 minutes ")
  stream.push("second 5 minutes ")
  stream.push("last 5 minutes")
})

Stream Consumer - I

To get the song, we first need to consume the Enumerator, and then flatMap the async response.

scala> val consumeSong = songStream(Iteratee.consume[String]())
consumeSong: s.c.Future[p.a.l.i.Iteratee[String,String]] = s.c.i.Promise$DefaultPromise@6854ee8b

flatMap the consumed song which is a Future[X]

scala> import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.ExecutionContext.Implicits.global

scala> import scala.concurrent.Future
import scala.concurrent.Future

scala> val songThatCanBeListened :Future[String] = consumeSong.flatMap(chunk => chunk.run)
songThatCanBeListened: s.c.Future[String] = s.c.i.Promise$DefaultPromise@2a7fc3c5

scala> songThatCanBeListened.onSuccess {case song => println(s"playing a song : $song")}

scala> playing a song : first 5 minutessecond 5 minuteslast 5 minutes

Stream Consumer - II

The other way of consuming the stream(Enumerator) is with piping/sinking operator(I don't exactly know what Play apis calls it),

scala> val consumeSongOtherWay = Iteratee.flatten(songStream |>> Iteratee.consume[String]()).run
consumeSongOtherWay: s.c.Future[String] = s.c.i.Promise$DefaultPromise@6c9f484a

scala> consumeSongOtherWay.onSuccess{case wholeSong => println(wholeSong)}

References

2.6.x/Enumerators

2.6.x/Iteratees

http://mandubian.com/2012/08/27/understanding-play2-iteratees-for-normal-humans/

Tweets about reactive scala

playing with docker native 1.12 container for mac

The following post might look like running kafka cluster in MacOS but it basically shows how to get started with Docker and basic docker commands.

Docker is widely being used to simply spawn a service I need as a separate container just like in production. And It can easily be guessed same container can be run the production server with the same config.

For example, if my application uses MongoDB, traditionally I used to download and install MongoDB myself and test it. I remember in one of my analytics project, I had weird issue with running different version of Elasticsearch because each Software Engineer does his own stuff in a large team.

With docker, I can create an Elasticsearch container that my application would connect to and share the same container with other members in a team. And the same Elasticsearch container is spawn inside the Production server, so that configuration everywhere is the same.

Download and install docker native for mac

wget https://download.docker.com/mac/stable/Docker.dmg
hdiutil mount Docker.dmg ##pops up installer to cp into /Applications
sudo cp -r /Volumes/Docker/Docker.app /Applications/
hdiutil unmount /Volumes/Docker

I will see following files in /Applications/

$ ls -l /Applications/Docker.app/Contents/
total 32
drwxr-xr-x  21 a1353612  admin   714 Oct  6 04:45 Frameworks
-rw-r--r--   1 a1353612  admin  2460 Oct  6 04:48 Info.plist
drwxr-xr-x   4 a1353612  admin   136 Oct  6 04:45 Library
drwxr-xr-x  11 a1353612  admin   374 Oct  6 04:48 MacOS
-rw-r--r--   1 a1353612  admin     8 Oct  6 04:45 PkgInfo
drwxr-xr-x  26 a1353612  admin   884 Oct  6 04:48 Resources
drwxr-xr-x   3 a1353612  admin   102 Oct  6 04:45 _CodeSignature
-rw-r--r--   1 a1353612  admin  7579 Oct  6 04:45 embedded.provisionprofile

Check its properly installed

docker --version
Docker version 1.12.1, build 6f9534c

docker-compose --version
docker-compose version 1.8.0, build f3628c7

docker-machine --version
docker-machine version 0.8.1, build 41b3b25

Test a webserver container is accessible from local.

docker run -it --rm -p 8888:8080 tomcat:8.0

Then send a GET request to localhost:8888

$ curl -X GET localhost:8888 -i
HTTP/1.1 200 OK
Server: Apache-Coyote/1.1
Content-Type: text/html;charset=UTF-8
Transfer-Encoding: chunked
Date: Wed, 22 Mar 2017 10:44:35 GMT

running a docker container

To run a docker container, I am cloning a kafka container, Kafka is a distributed messaging/streaming platform.

git clone https://github.com/prayagupd/docker-kafka-zk.git
cd docker-kafka-zk
docker-compose up -d

docker-compose is a tool for defining and running multi-container Docker apps.

Check the running containers

docker ps
CONTAINER ID        IMAGE                    COMMAND                  CREATED             STATUS              PORTS                                                NAMES
4369c261f6d4        wurstmeister/zookeeper   "/bin/sh -c '/usr/sbi"   48 seconds ago      Up 45 seconds       22/tcp, 2888/tcp, 3888/tcp, 0.0.0.0:2181->2181/tcp   dockerkafkazk_zookeeper_1
6e83021e6c97        dockerkafkazk_kafka      "start-kafka.sh"         48 seconds ago      Up 46 seconds       0.0.0.0:32768->9092/tcp                              dockerkafkazk_kafka_1

Get the IP addresses for the containers

docker exec dockerkafkazk_kafka_1 cat /etc/hosts
127.0.0.1 localhost
::1 localhost ip6-localhost ip6-loopback
fe00::0 ip6-localnet
ff00::0 ip6-mcastprefix
ff02::1 ip6-allnodes
ff02::2 ip6-allrouters
172.18.0.2 6e83021e6c97

or 

docker inspect dockerkafkazk_kafka_1 | grep -w "IPAddress"
            "IPAddress": "",
                    "IPAddress": "172.18.0.2",

and for zookeeper, 

docker exec dockerkafkazk_zookeeper_1 cat /etc/hosts
127.0.0.1 localhost
::1 localhost ip6-localhost ip6-loopback
fe00::0 ip6-localnet
ff00::0 ip6-mcastprefix
ff02::1 ip6-allnodes
ff02::2 ip6-allrouters
172.18.0.3 4369c261f6d4

SecureSHell to Kafka container from local machine

./start-kafka-shell.sh 172.18.0.2 172.18.0.3:2181
bash-4.3# echo $KAFKA_HOME/
/opt/kafka_2.11-0.10.0.1/

Fire the following command to stop all containers,

docker stop $(docker ps -a -q)

In next post, I will probably write about streaming to this kafka container using nodejs. I'm currently doing some stuffs on nodejs during my Masters period, (off office work). So, thinking of streaming via nodejs app which scala app would consume on the other side.

Resources
--------------

https://www.viget.com/articles/how-to-use-docker-on-os-x-the-missing-guide

https://docs.docker.com/docker-for-mac/networking/

Virtualization 101

What is virtualization?

The idea of virtualization is to to be able to run virtual computers on existing piece of physical machine. One computer will get one IP address, with virtualization I can have multiple IPs on same hardware.

I first knew of virtualization while working on software for banks. The "IT department" of bank would create a VM for me to install software :)

https://www.redhat.com/en/topics/virtualization/what-is-virtualization

Setting up VM

Install VirtualBox which is a full virtualizer https://www.virtualbox.org/wiki/Downloads

Then download the ISO for OS you want to have a VM for and attach it to VirtualBox storage as shown in screenshot below(In MacOS). Ubuntu ISO can be downloaded from here, https://www.ubuntu.com/download/desktop

http://askubuntu.com/questions/825437/ubuntu-16-in-vm-does-not-have-access-to-internet

Regarding the Network setup, it really depends on the nature of the network being used. In following example, in one case wireless works for me.

while in wired environment ethernet adaptor works.

Resizing the VM
In case you need to increase the VM RAM or Disk size, after you created your VM, then follow the process as below :

STEP 1 : shut down VM

STEP 2 : resize the VM vdi using host terminal (windows in mycase)
VBoxManage modifyhd "C:\Users\AS18\VirtualBox VMs\WM_Bridge\WM_Bridge.vdi" --resize 100000

STEP 3 : 
Then, using gparted resize the /dev/sda1

Original config

Delete linux-swap

Reassign the size to linux-swap

http://askubuntu.com/a/558215/37643

http://askubuntu.com/a/95026/37643

Share folder
---------------------

1) Install guest edition

followed by sudo /media/prayagupd/VBOXADDITIONS_5.1.8_111374/VBoxLinuxAdditions.run and then reboot.

How to Access Folders on Your Host Machine from an Ubuntu Virtual Machine in VirtualBox

2) add current guest user to the vboxsf usergroup

sudo adduser $USER vboxsf
[sudo] password for prayagupd:
Adding user `prayagupd' to group `vboxsf' ...
Adding user prayagupd to group vboxsf
Done.

verify that its working, 

id $USER

uid=1000(prayagupd) gid=1000(prayagupd) groups=1000(prayagupd),4(adm),24(cdrom),27(sudo),30(dip),46(plugdev),113(lpadmin),128(sambashare),999(vboxsf)

3) mount shared folder

Create a mount folder on host machine with Auto-mount Yes, and Access Full.

use flume to stream from avro source and publish to logger sink

flume = a deep narrow channel or ravine with a stream running through it. 

PART 1 - setup flume (can be dockerized)

wget --proxy=off http://apache.claz.org/flume/1.6.0/apache-flume-1.6.0-bin.tar.gz
tar -zxvf apache-flume-1.5.0-bin.tar.gz -C /opt/flume --strip-components=1
rm apache-flume-1.6.0-bin.tar
sudo chmod -R 777 /opt/flume

PART 2 start the flume node (in a docker container)

conf/supply_chain_flume.conf

# Define a memory channel called ch1 on supply_agent                                                        
supply_agent.channels.logEventStream.type = memory

# Define an Avro source called avro-source1 on agent1 and tell it                                     
# to bind to 0.0.0.0:41414. Connect it to channel ch1.                                               
supply_agent.sources.avro-source1.channels = logEventStream
supply_agent.sources.avro-source1.type = avro
supply_agent.sources.avro-source1.bind = 0.0.0.0
supply_agent.sources.avro-source1.port = 41414

# Define a logger sink that simply logs all events it receives                                       
# and connect it to the other end of the same channel.                                               
supply_agent.sinks.log-sink1.channel = logEventStream
supply_agent.sinks.log-sink1.type = logger

# Finally, now that we've defined all of our components, tell                                         
# agent1 which ones we want to activate.                                                           
supply_agent.channels = logEventStream
supply_agent.sources = avro-source1
supply_agent.sinks = log-sink1

start flume agent

bin/flume-ng agent --conf ./conf/ -f conf/supply_chain_flume.conf -Dflume.root.logger=DEBUG,console -n supply_agent

Info: Sourcing environment configuration script /usr/local/apache-flume-1.6.0-bin/conf/flume-env.sh
+ exec /Library/Java/JavaVirtualMachines/jdk1.8.0_65.jdk/Contents/Home//bin/java -Xmx20m -Dflume.root.logger=DEBUG,console -cp '/usr/local/apache-flume-1.6.0-bin/conf:/usr/local/apache-flume-1.6.0-bin/lib/*' -Djava.library.path= org.apache.flume.node.Application -f conf/flume.conf -n supply_agent
2016-09-22 00:08:49,909 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.node.PollingPropertiesFileConfigurationProvider.start(PollingPropertiesFileConfigurationProvider.java:61)] Configuration provider starting
2016-09-22 00:08:49,912 (lifecycleSupervisor-1-0) [DEBUG - org.apache.flume.node.PollingPropertiesFileConfigurationProvider.start(PollingPropertiesFileConfigurationProvider.java:78)] Configuration provider started
2016-09-22 00:08:49,913 (conf-file-poller-0) [DEBUG - org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:126)] Checking file:conf/flume.conf for changes
2016-09-22 00:08:49,914 (conf-file-poller-0) [INFO - org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:133)] Reloading configuration file:conf/flume.conf
2016-09-22 00:08:49,918 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)] Processing:log-sink1
2016-09-22 00:08:49,918 (conf-file-poller-0) [DEBUG - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1021)] Created context for log-sink1: channel
2016-09-22 00:08:49,918 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:931)] Added sinks: log-sink1 Agent: supply_agent
2016-09-22 00:08:49,918 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)] Processing:log-sink1
2016-09-22 00:08:49,919 (conf-file-poller-0) [DEBUG - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:314)] Starting validation of configuration for agent: supply_agent, initial-configuration: AgentConfiguration[supply_agent]
SOURCES: {avro-source1={ parameters:{bind=0.0.0.0, channels=ch1, port=41414, type=avro} }}
CHANNELS: {ch1={ parameters:{type=memory} }}
SINKS: {log-sink1={ parameters:{channel=ch1, type=logger} }}

2016-09-22 00:08:49,922 (conf-file-poller-0) [DEBUG - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateChannels(FlumeConfiguration.java:469)] Created channel ch1
2016-09-22 00:08:49,926 (conf-file-poller-0) [DEBUG - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateSinks(FlumeConfiguration.java:675)] Creating sink: log-sink1 using LOGGER
2016-09-22 00:08:49,927 (conf-file-poller-0) [DEBUG - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:372)] Post validation configuration for supply_agent
AgentConfiguration created without Configuration stubs for which only basic syntactical validation was performed[supply_agent]
SOURCES: {avro-source1={ parameters:{bind=0.0.0.0, channels=ch1, port=41414, type=avro} }}
CHANNELS: {ch1={ parameters:{type=memory} }}
AgentConfiguration created with Configuration stubs for which full validation was performed[supply_agent]
SINKS: {log-sink1=ComponentConfiguration[log-sink1]
  CONFIG:
    CHANNEL:ch1
}

2016-09-22 00:08:49,927 (conf-file-poller-0) [DEBUG - org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:136)] Channels:ch1

2016-09-22 00:08:49,927 (conf-file-poller-0) [DEBUG - org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:137)] Sinks log-sink1

2016-09-22 00:08:49,927 (conf-file-poller-0) [DEBUG - org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:138)] Sources avro-source1

2016-09-22 00:08:49,927 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:141)] Post-validation flume configuration contains configuration for agents: [supply_agent]
2016-09-22 00:08:49,928 (conf-file-poller-0) [INFO - org.apache.flume.node.AbstractConfigurationProvider.loadChannels(AbstractConfigurationProvider.java:145)] Creating channels
2016-09-22 00:08:49,933 (conf-file-poller-0) [INFO - org.apache.flume.channel.DefaultChannelFactory.create(DefaultChannelFactory.java:42)] Creating instance of channel ch1 type memory
2016-09-22 00:08:49,936 (conf-file-poller-0) [INFO - org.apache.flume.node.AbstractConfigurationProvider.loadChannels(AbstractConfigurationProvider.java:200)] Created channel ch1
2016-09-22 00:08:49,937 (conf-file-poller-0) [INFO - org.apache.flume.source.DefaultSourceFactory.create(DefaultSourceFactory.java:41)] Creating instance of source avro-source1, type avro
2016-09-22 00:08:49,953 (conf-file-poller-0) [INFO - org.apache.flume.sink.DefaultSinkFactory.create(DefaultSinkFactory.java:42)] Creating instance of sink: log-sink1, type: logger
2016-09-22 00:08:49,955 (conf-file-poller-0) [INFO - org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:114)] Channel ch1 connected to [avro-source1, log-sink1]
2016-09-22 00:08:49,960 (conf-file-poller-0) [INFO - org.apache.flume.node.Application.startAllComponents(Application.java:138)] Starting new configuration:{ sourceRunners:{avro-source1=EventDrivenSourceRunner: { source:Avro source avro-source1: { bindAddress: 0.0.0.0, port: 41414 } }} sinkRunners:{log-sink1=SinkRunner: { policy:org.apache.flume.sink.DefaultSinkProcessor@2d9e85b0 counterGroup:{ name:null counters:{} } }} channels:{ch1=org.apache.flume.channel.MemoryChannel{name: ch1}} }
2016-09-22 00:08:49,969 (conf-file-poller-0) [INFO - org.apache.flume.node.Application.startAllComponents(Application.java:145)] Starting Channel ch1
2016-09-22 00:08:50,021 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.register(MonitoredCounterGroup.java:120)] Monitored counter group for type: CHANNEL, name: ch1: Successfully registered new MBean.
2016-09-22 00:08:50,021 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.start(MonitoredCounterGroup.java:96)] Component type: CHANNEL, name: ch1 started
2016-09-22 00:08:50,021 (conf-file-poller-0) [INFO - org.apache.flume.node.Application.startAllComponents(Application.java:173)] Starting Sink log-sink1
2016-09-22 00:08:50,022 (conf-file-poller-0) [INFO - org.apache.flume.node.Application.startAllComponents(Application.java:184)] Starting Source avro-source1
2016-09-22 00:08:50,022 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.source.AvroSource.start(AvroSource.java:228)] Starting Avro source avro-source1: { bindAddress: 0.0.0.0, port: 41414 }...
2016-09-22 00:08:50,023 (SinkRunner-PollingRunner-DefaultSinkProcessor) [DEBUG - org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:143)] Polling sink runner starting
2016-09-22 00:08:50,254 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.register(MonitoredCounterGroup.java:120)] Monitored counter group for type: SOURCE, name: avro-source1: Successfully registered new MBean.
2016-09-22 00:08:50,254 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.start(MonitoredCounterGroup.java:96)] Component type: SOURCE, name: avro-source1 started
2016-09-22 00:08:50,254 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.source.AvroSource.start(AvroSource.java:253)] Avro source avro-source1 started.


2016-09-22 00:09:50,023 (conf-file-poller-0) [DEBUG - org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:126)] Checking file:conf/flume.conf for changes




PART 3 - publish to avro source listening on port 41414

Running an Avro client, sends either a file or data from stdin to a specified host and port where a Flume NG Avro Source is listening.

/var/log/supply_source.log

{"timemillis" : 2876873673, "correlation" : 1, item : "pants"}
{"timemillis" : 8347583748, "correlation" : 2, item : "shirts"}

bin/flume-ng avro-client --conf conf -H localhost -p 41414 -F /var/log/supply_source.log -Dflume.root.logger=DEBUG,console

Now, the sink receives two events.

2016-09-22 00:36:36,610 (New I/O server boss #1 ([id: 0x4ca942fa, /0:0:0:0:0:0:0:0:41414])) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:171)] [id: 0x49d18056, /127.0.0.1:50833 => /127.0.0.1:41414] OPEN
2016-09-22 00:36:36,610 (New I/O  worker #4) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:171)] [id: 0x49d18056, /127.0.0.1:50833 => /127.0.0.1:41414] BOUND: /127.0.0.1:41414
2016-09-22 00:36:36,610 (New I/O  worker #4) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:171)] [id: 0x49d18056, /127.0.0.1:50833 => /127.0.0.1:41414] CONNECTED: /127.0.0.1:50833
2016-09-22 00:36:36,802 (New I/O  worker #4) [DEBUG - org.apache.flume.source.AvroSource.appendBatch(AvroSource.java:371)] Avro source avro-source1: Received avro event batch of 2 events.

2016-09-22 00:36:36,802 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{} body: 7B 22 63 6F 72 72 65 6C 61 74 69 6F 6E 22 20 3A {"correlation" : }
2016-09-22 00:36:36,802 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{} body: 7B 22 63 6F 72 72 65 6C 61 74 69 6F 6E 22 20 3A {"correlation" : }
2016-09-22 00:36:36,817 (New I/O  worker #4) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:171)] [id: 0x49d18056, /127.0.0.1:50833 :> /127.0.0.1:41414] DISCONNECTED
2016-09-22 00:36:36,817 (New I/O  worker #4) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:171)] [id: 0x49d18056, /127.0.0.1:50833 :> /127.0.0.1:41414] UNBOUND
2016-09-22 00:36:36,817 (New I/O  worker #4) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:171)] [id: 0x49d18056, /127.0.0.1:50833 :> /127.0.0.1:41414] CLOSED

2016-09-22 00:36:36,818 (New I/O  worker #4) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.channelClosed(NettyServer.java:209)] Connection to /127.0.0.1:50833 disconnected.

Reference

docker flume example, https://github.com/prayagupd/docker-flume

Parallel application using OpenMP/ MacOS

brew install clang-omp
xcode-select --install

$ clang-omp --version
clang version 3.5.0
Target: x86_64-apple-darwin15.4.0
Thread model: posix

$ xcode-select --version
xcode-select version 2343.

application

#include <omp.h>
#include <stdio.h>

#define n 20                                                                                          

omp_set_num_threads(3);                                                                              
#pragma omp parallel for private(tid) schedule(static,1)                                              

for (i=0; i<n; i++) {                                                                                
 tid = omp_get_thread_num();                                                                          
 printf("Thread %d executing iteration %d\n", tid, i);                                                
}

$ clang-omp -fopenmp parallel.c 

Mac OS - command line hacks

processor bits

"processor bits" means that how much data a microprocessor will process within one instruction cycle i.e. fetch-decode-execute

32-bit means that a microprocessor can execute 4 bytes of data in one instruction cycle

while 64-bit means that a microprocessor executes can execute 8 bytes of data in one instruction cycle.

$ sysctl hw | grep cpu #sysctl hw.cpu64bit_capable
hw.ncpu: 8
hw.activecpu: 8
hw.physicalcpu: 4
hw.physicalcpu_max: 4
hw.logicalcpu: 8
hw.logicalcpu_max: 8
hw.cputype: 7
hw.cpusubtype: 8
hw.cpu64bit_capable: 1
hw.cpufamily: 280134364
hw.cpufrequency: 2800000000
hw.cpufrequency_min: 2800000000
hw.cpufrequency_max: 2800000000
hw.cputhreadtype: 1

$ getconf LONG_BIT
64

The number of bits in a processor refers to the size of the data types that it handles and the size of its registry. Simply put, a 64-bit processor is more capable than a 32-bit processor because it can handle more data at once.

CPU info

sysctl -n machdep.cpu.brand_string
Intel(R) Core(TM) i7-4980HQ CPU @ 2.80GHz

$ system_profiler | grep Processor
    Apple: AUDynamicsProcessor:
    Apple: AUVoiceProcessor:
      Processor Name: Intel Core i7
      Processor Speed: 2.8 GHz
      Number of Processors: 1
    <key>BootCampProcessorPstates</key>
    |         "BootCampProcessorPstates" = <0a00>

no of cores

$ sysctl -n hw.ncpu
8

IP address of Machine

$ ipconfig getifaddr en0
10.11.208.214

or

$ ifconfig | grep inet
inet6 ::1 prefixlen 128
inet 127.0.0.1 netmask 0xff000000
inet6 fe80::1%lo0 prefixlen 64 scopeid 0x1
inet6 fe80::c6b3:1ff:fed3:d771%en0 prefixlen 64 scopeid 0x4
inet 10.11.208.214 netmask 0xfffffe00 broadcast 10.11.209.255
inet6 fe80::7023:7bff:fee3:61e2%awdl0 prefixlen 64 scopeid 0x9

Change password using DsCl (Directory Service CLI)

dscl . -passwd /Users/prayagupd
New Password: dreams
passwd: DS error: eDSAuthPasswordTooShort

<dscl_cmd> DS Error: -14170 (eDSAuthPasswordTooShort)

Note

$ man dscl | grep "passwd"

     Usage: passwd user_path [new_pasword | old_password new_pasword]

Uninstall application

mdfind -name "Skype"
/Applications/Skype.app

sudo rm -rf /Applications/Skype.app

add a printer (lpadmin)

find the IP address of the printer, and then add it to macos printers.

http://apple.stackexchange.com/q/126834/76240


printer stat

lpstat -p
printer _10_10_10_113 is idle.  enabled since Thu Oct 20 17:23:31 2016
 Waiting for printer to finish.
printer _10_20_50_100 is idle.  enabled since Fri Sep  9 15:39:45 2016

RAM Memory usage

$ vm_stat 
Mach Virtual Memory Statistics: (page size of 4096 bytes)
Pages free:                              264375.
Pages active:                           2027888.
Pages inactive:                          651659.
Pages speculative:                        23565.
Pages throttled:                              0.
Pages wired down:                        852647.
Pages purgeable:                          39328.
"Translation faults":                 880109291.
Pages copy-on-write:                   44871249.
Pages zero filled:                    433518664.
Pages reactivated:                      5346396.
Pages purged:                           1423915.
File-backed pages:                       379668.
Anonymous pages:                        2323444.
Pages stored in compressor:             2870604.
Pages occupied by compressor:            372004.
Decompressions:                        33640670.
Compressions:                          43459639.
Pageins:                               10427527.
Pageouts:                                 34549.
Swapins:                               32669611.
Swapouts:                              34330610.