20 Jul 2016

Scala: Pattern matching & PartialFunction

Scala cung cấp tính năng pattern matching (và được sử dụng rất nhiều), tương ứng với switch của Java. Nhưng syntax có khác đi một chút.


Trong Java chúng ta dùng switch như  bên dưới:
String month;
switch (myNumber) {
  case 1:  month = "thang 1";
  break;
  case 2: month = "thang 2";
  break;
  case 3:
  case 4:
    default: month = "xxxx";
}

return month;
General syntax của swicth trong Java có dạng switch (selector) { alternatives } (1).
Đối với switch của Java, khi dừng câu lệnh switch, chúng ta phải dùng break, nếu không, khi gặp pattern được match, switch sẽ "fall through" đến các case tiếp theo cho đến khi nào gặp break, rất dễ xảy ra lỗi nếu như chúng ta không để ý, và IDE cũng không warning trong trường hợp có return type hợp lệ.


Dùng scala, khi sử dụng pattern matching, syntax sẽ là selector match { alternatives }

Pattern matching của scala cũng sử dụng từ khoá case như Java, tiếp theo là pattern muốn match, và dấu => để tách biệt pattern và biểu thức dùng để xử lí khi pattern được match. match trong scala khác biệt so với Java:
1. đầu tiên do pattern matching của Scala không "fall through" đến các case matching khác, nên không cần dùng break
2. Mỗi một biểu thức alternative trong scala đều phải sinh ra giá trị, cho dù rơi vào default case 
3. Đối với Java, nếu input không rơi vào case nào (không được match) và không có case default, sẽ không có lỗi xảy ra, nhưng có thể sẽ sinh ra lỗi về sau. Đối với Scala, nếu sử dụng pattern matching, mỗi khi dùng match chúng ta buộc phải return giá trị, dù input không match bất cứ case nào đi nữa, nếu không sẽ sinh ra exception MatchError.

Với ví dụ trên, khi dùng với scala sẽ như sau:

val monthName: String = month match {
  case 1 => "thang 1"  case 2 => "thang 2"  case _ => "xxx"}

Pattern matching cuả scala có thể dùng với nhiều dạng pattern: wildcard, constant (như ví dụ trên, 1, 2 là constant, _ là wildcard), case class, type, tuple, sequence, constructor...

Do vậy chúng ta hoàn toàn có thể sử dụng các cách thức sau đối với pattern matching:

case 1 => "thang 1"
hay là đối với list
case _ :: y :: _ if y > 1 => y
Với type
case s: String => s.length
v.v    

Một vấn đề khi sử dụng pattern matching là chúng ta thường không kiểm soát được input, và compiler không có cách nào để thông báo khi chúng ta dùng logic sai, một ví dụ đối với if else thông thường trong scala:

def myFunc(seq : List[Int]): Int = {
  if (seq.length > 1) seq(2)
  else 0}



Khi sử dụng myFunc với input là List Int có length nhỏ hơn 1 hoặc lớn hơn 3 sẽ cho kết quả đúng, nhưng nếu như input là một List với length bằng 2 chương trình sẽ crash:

scala> def myFunc(seq : List[Int]): Int = {
  |     if (seq.length > 1) seq(2)
  |     else 0  |   }
myFunc: (seq: List[Int])Int

scala> myFunc(List(2, 3, 4))
res6: Int = 4
scala> myFunc(List(2, 4))
java.lang.IndexOutOfBoundsException: 2at scala.collection.LinearSeqOptimized$class.apply(LinearSeqOptimized.scala:65)
at scala.collection.immutable.List.apply(List.scala:84)
at .myFunc(<console>:12)  ... 32 elided

Tương tự nếu sử dụng pattern matching mà chúng ta không cover hết tất cả các case, sẽ gặp phải Runtime exception. 

scala> val second: List[Int] => Int = { case _ :: y :: _  if y > 1 => y }
second: List[Int] => Int = <function1>
scala> second(List(1, 2))res12: Int = 2
scala> second(List(1))scala.MatchError: List(1) 
(of class scala.collection.immutable.$colon$colon)at 
$anonfun$1.apply(<console>:11)  
at $anonfun$1.apply(<console>:11)    ... 32 elided


Để khắc phục lỗi này, scala cung cấp PartialFunction.
val second: PartialFunction[List[Int],Int] = {
  case _ :: y :: _  if y > 1 => y
}

Hàm trên khi scala compile sẽ được dịch ra tương tự như 

val second = new PartialFunction[List[Int], Int]  {
  def apply(xs: List[Int]) = xs match {
    case _ :: y :: _  if y > 1 => y
  }

  def isDefinedAt(xs: List[Int]) = xs match {
    case _ :: y :: _ => true    
    case _ => false  
  }
}


Như vậy Partial function cung cấp một method dùng để kiểm tra function matcher của chúng ta có làm việc được với input cụ thể nào đó không. Nếu như không kiểm tra mà cứ dùng như ví dụ ban đầu, chúng ta vẫn sẽ gặp Runtime exception, do vậy khi define function matcher với Partial Function, thì lúc sử dụng, chúng ta dùng method isDefinedAt để kiểm tra giá trị input trước:

scala> val second: PartialFunction[List[Int],Int] = {
    case _ :: y :: _  if y > 1 => y
   }
second: PartialFunction[List[Int],Int] = <function1>

scala>   val l = 1 :: 2 :: 3 :: Nil
l: List[Int] = List(1, 2, 3)

scala>

scala>   val a = if (second.isDefinedAt(l)) second(l) else 0
a: Int = 2
scala>

scala>   val l1 = 1 :: Nil
l1: List[Int] = List(1)

scala>

scala>   val b = if (second.isDefinedAt(l1)) second(l) else 0
b: Int = 0

Ứng dụng:

Scala cung cấp các method để làm việc trên scala standard collection như Seq, List, Set... Đối với các method như filter, map chấp nhận input là các Anonymous function với parameters phù hợp để sinh ra các collection mới theo tiêu chí đặt ra trong Anonymous function, ví dụ khi làm việc với một List[Int] đơn giản như bên dưới:

val l = 1 :: 2 :: 3 :: Nil

Có thể dùng map và filter như sau:

scala> l.map({ case x => x*2 })
res7: List[Int] = List(2, 4, 6)

scala> l.filter(_ > 1)
res8: List[Int] = List(2, 3)

Để ý cấu trúc map method bên trên, ta sử dụng một Anonymous function dùng pattern matching là bất kì input nào, do đó tất cả các case đều được tính đến. Nhưng nếu như có nhu cầu chỉ lấy ra một List các số Int lớn hơn 1 trong List ban đầu, khi dùng map sẽ gặp lỗi:

scala> l.map({ case x if x > 1 => x })
scala.MatchError: 1 (of class java.lang.Integer)
at $anonfun$1.apply$mcII$sp(<console>:13)  
at $anonfun$1.apply(<console>:13)    
at $anonfun$1.apply(<console>:13)      
at scala.collection.immutable.List.map
(List.scala:273)      
... 32 elided

Vì sự khác nhau giữa switch của Java và match của Scala như đã nói từ đầu bài, tất cả các case đều cần được tính đến. Phần matching chúng ta dùng Pattern guard (if x > 1) để lọc giá trị lớn hơn 1, nhưng chưa match giá trị bằng hoặc nhỏ hơn 1, do vậy sẽ gặp MatchError Exception.

Để giải quyết vấn đề này, có thể dùng collect method với Partial Function:
scala> val moreThanOne: PartialFunction[Int,Int] = { case x if x > 1 => x }
moreThanOne: PartialFunction[Int,Int] = <function1>
scala> l.collect(moreThanOne)
res12: List[Int] = List(2, 3)

Như vậy với Partial Function chúng ta có thể giải quyết bài toán này theo một cách cực kì đơn giản.

 (1) Programming in Scala 2nd Edition

26 Feb 2016

Code review

This's let's Encrypt, where people contribute their code to build a simple, free, great thing that most sysadmin, web developer, companies really want, free ssl certificate for all.

Look at how the review a small piece of code:

https://github.com/letsencrypt/letsencrypt/pull/2453#issuecomment-189229778

That's just so great, coding style, compatibility concern, test case, follow rfc and more.

5 Mar 2015

Something I have done these days

1. Symemcache pool using common pool 1.x: https://github.com/whatvn/Spymemcache-commonpool-1 
2. Symemcache pool using common pool 2.x: https://github.com/whatvn/Spymemcache-commonpool-2 
3. A http multiplexer in java: https://github.com/whatvn/HttpMultiplexer
4. A video encoder service which makes use of libav (from fffmpeg), gearman and libftp: https://github.com/whatvn/mp4encoder
5. An example nginx module to make nginx work together with thrift backend: https://github.com/whatvn/nginx_photo_thrift_module 
6. A modified version of nginx mp4 module to fix mp4 file automatically in order to start playback immediately: https://github.com/whatvn/ngx_http_enhance_mp4_module 
7. A ported version of firebase pushid in C (can be called a better/simpler uuid generator): https://github.com/whatvn/firebase-pushid 

You can find more details in github links. 


19 Aug 2014

Nutch: crawl URL from kafka


Apache Nutch is a java framework, used to build a distributed, multiprocessing parser and crawler, it uses map/reduce engine to distribute work across multi node for fast parsing and indexing web pages. 

Many people use Nutch will get familiar with this software by using its shell scripts (nutch and crawl script) which is shipped along with Nutch source code. 

When start using this project, I found it's difficult to integrate project's source code into our application, because we did not use anything like Spring framework with xml configuration before, other than that, Nutch comes with many hadoop's style configuration files, and these file has to be path of project in order to get application working. 

1. Follow steps in Apache Nutch's wiki   (install hbase, modify configuration, compile using ant). 
2. Add all compiled's library to Netbeans project.
3. Add modified configuration files as part of project.





4. And then work with Nutch source code.

The default Nutch's behaviour is taking list of URL from a special local folder, and passing it to Crawler class as Nutch.ARG_SEEDDIR which is somewhat not suitable for most of us, we often do not want to take a list of URLs, then create a file with these content manually and pass the directory into Nutch as seedDir argument. 

When looking into Nutch source code, it has another option, get a list URLs as argument and create a HDFS directory, it does not have to be a seedDir:
 String crawlId = (String) args.get(Nutch.ARG_CRAWL);
        if (crawlId != null) {
            getConf().set(Nutch.CRAWL_ID_KEY, crawlId);
        }
        String seedDir = null;
        String seedList = (String) args.get(Nutch.ARG_SEEDLIST);
        if (seedList != null) { // takes precedence
            String[] seeds = seedList.split("\\s+");
            // create tmp. dir
            String tmpSeedDir = getConf().get("hadoop.tmp.dir") + "/seed-"
                    + System.currentTimeMillis();
            FileSystem fs = FileSystem.get(getConf());
            Path p = new Path(tmpSeedDir);
            fs.mkdirs(p);
            Path seedOut = new Path(p, "urls"); 
           OutputStream os = fs.create(seedOut);
            for (String s : seeds) {
                os.write(s.getBytes());
                os.write('\n');
            }
            os.flush();
            os.close();
            cleanSeedDir = true;
            seedDir = tmpSeedDir;
        } else {
            seedDir = (String) args.get(Nutch.ARG_SEEDDIR);
        }

What we are going to do is simple, when starting our crawler, just give Nutch seedList as a Java String separated by " " instead of seedDir

String seedUrl = "www.google.com www.blogger.com";                
_logger.info("Processing url [{}]", seedUrl);                
Random random = new Random();                
String batchId = String.valueOf(random.nextInt());                
Configuration nutchConfiguration = NutchConfiguration.create();                
nutchConfiguration.set(BATCH_ID, batchId);                
String solrUrl = nutchConfiguration.get(SOLR_URL);                
String crawlArgs = String.format("-seedurl %s -depth 5 -topN 10", seedUrl);                
// Run Crawl tool                
ToolRunner.run(nutchConfiguration, new Crawler(),                        
            (crawlArgs));

It's not going to work. As previous source code showed above, seedList will take precedence (instead of seedDir), but next phase (InjectorJob) won't get seedList argument, it takes seedDir (which is a mistake?)


public Map<String,Object> run(Map<String,Object> args) throws Exception {    
    getConf().setLong("injector.current.time", System.currentTimeMillis());    
    Path input;    
    Object path = args.get(Nutch.ARG_SEEDDIR);    
    if (path instanceof Path) {      
        input = (Path)path;    
    } else {      
    input = new Path(path.toString()); 
       }    
....}


To get Nutch work with seedList, just put an seedDir argument into argument map in Crawler step:

 if (seedList != null) { // takes precedence
            String[] seeds = seedList.split("\\s+");
            // create tmp. dir
            String tmpSeedDir = getConf().get("hadoop.tmp.dir") + "/seed-"
                    + System.currentTimeMillis();
            FileSystem fs = FileSystem.get(getConf());
            Path p = new Path(tmpSeedDir);
            fs.mkdirs(p);
            Path seedOut = new Path(p, "urls"); 
           OutputStream os = fs.create(seedOut);
            for (String s : seeds) {
                os.write(s.getBytes());
                os.write('\n');
            }
            os.flush();
            os.close();
            cleanSeedDir = true;
            seedDir = tmpSeedDir;
            args.put(Nutch.ARG_SEEDDIR, seedDir);
        } else {
            seedDir = (String) args.get(Nutch.ARG_SEEDDIR);
        }

 and everything will be going well. 
Last step, get url from kafka, listen for a special kafka topic, get the message and pass it as seedList argument:

 public void process(String topic, ConsumerConnector consumer) {

        Map<String, Integer> topicCountMap = new HashMap<>();
        topicCountMap.put(topic, new Integer(1));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
        KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
        ConsumerIterator<byte[], byte[]> it = stream.iterator();

        while (it.hasNext()) {
            try {
                MessageAndMetadata<byte[], byte[]> next = it.next();
                String seedUrl = new String(next.message());
                _logger.info("Processing url [{}]", seedUrl);
                Random random = new Random();
                String batchId = String.valueOf(random.nextInt());
                Configuration nutchConfiguration = NutchConfiguration.create();
                nutchConfiguration.set(BATCH_ID, batchId);
                String solrUrl = nutchConfiguration.get(SOLR_URL);
                String crawlArgs = String.format("-seedurl %s -depth 5 -topN 10", seedUrl);
                // Run Crawl tool
                ToolRunner.run(nutchConfiguration, new Crawler(),
                        tokenize(crawlArgs));
                SolrIndexerJob solrIndexerJob = new SolrIndexerJob();
                solrIndexerJob.setConf(nutchConfiguration);
                solrIndexerJob.indexSolr(solrUrl, batchId);
            } catch (Exception ex) {
                java.util.logging.Logger.getLogger(ABCrawler.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }
Happy crawling!

20 Sept 2012

Re-implement thing


This is re-implemented version of rbalancer in pure python without using any framework and gevent. It works exactly the way rbalancer do:https://github.com/whatvn/rbalancer and work really better.

Please check it out at https://github.com/whatvn/py-balancer if you wanna try ;). 

22 Jul 2012

Tsar

If you've been working with Linux system for enough time, you will know sar. Since sar supports collecting many system statistic and its syntax is a bit complicated, Yahoo people created a tool in Perl called Ysar, it's similar to sar with simpler syntax, Yahoo has not published that tool yet. People at Taobao created another tool in C, named it with tsar and open source it at: https://gitorious.org/trafficserver/tsar/commits/master


Working with Apache Traffic server to build internal CDN for our company, I also join #trafficserver on FreeNode, that's when I heard about Tsar and started using it.  The opensource version of Tsar has 2 features, once is to get its syntax simpler and easier to use than sar, another is collecting various statistic of Apache Traffic server, integrate with nagios. For some reasons, this public version does not work in some cases: 

  • It only support traffic server installed as standard Red Hat package, so if you installed traffic server on another Linux distro,  BSD or Sun Solaris, it will not work. 
  • Module traffic does not work.
  • it  only supports standard network ethernet (eth*), bonding and p2p device won't work. 
I modified tsar to fix above problems. the modified version will work in any kind of Apache traffic server on any system. It support bonding and p2p and standard ethernet device....
My pull requests was not accepted by Taobao developer, so I put it on github  (Installation and Usage in README file). If you're using traffic server with nagios monitor system, this tool will fit your need. 


Load balancing using HTTP redirect reponse

This May I wrote a simple HTTP load balancer using HTTP 302 redirect response based on Facebook Tornado, it's called rbalancer  
After running it for 2 months, I'm confidence that it's stable and quite good for some purpose.
This's a short introduction of rbalancer:

rbalancer is a simple HTTP load balancer using HTTP 302 redirect response with round-robin and weighted random support built ontornadorbalancer can perform health check on server in balancer list, when a server went down, rbalancer will automatic re-balance request to alive servers. When dead one goes up, it will be added to cluster and serving request likes normal

If you're curious  about this thing, come check it on github

Disqus