Handling zip files in Google Cloud Dataflow

Google cloud dataflow is useful in processing large amounts of data but when it comes to processing the files in zip files, it is a bit tricky. By design, dataflow doesn't distribute the zip files from a bucket to multiple workers. Instead it allocates only one worker to unzip all the files and process them. This is not an optimal solution for the amount of data we need to handle.

There is an alternate way to do this:

1. Put all the zip file names in a text file. Note: The following model assumes all the zip files are in GCS path.
2. Give the text file name to the pipeline.
3. It will go through the bucket, read the zip file names and create the absolute GCS file path and outputs them to the next in the pipeline.
4. The next pipeline will take this GCS filepath and without caring about the file type, will unzip each input zip file and then hand over the contents to the next in the pipeline.
5. Now instead of one worker, you have N workers to handle the zip files

The code for assigning the absolute GCS path:

import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.util.GcsUtil;
import org.apache.beam.sdk.util.gcsfs.GcsPath;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.List;

public class GetGCSUtilPath extends DoFn<String, String> {
    private static final Log log = LogFactory.getLog(GetGCSUtilPath.class);

    @ProcessElement
    public void processElement(ProcessContext c){
        try {
            GcsUtil.GcsUtilFactory factory = new GcsUtil.GcsUtilFactory();
            GcsUtil util = factory.create(c.getPipelineOptions());
            String dataFolder = c.element();
            log.warn("Processing data folder - " + dataFolder);
            List<GcsPath> gcsPaths = util.expand(GcsPath.fromUri(dataFolder));

            for (GcsPath gcsp : gcsPaths) {
                c.output(gcsp.toString());
            }
        } catch(Exception ex) {
            log.error("Unalbe to process the file " + c.element());
        }
    }
}


JavaAutoApplescriptBashCClojureC++C SharpCSSElmGoGraphQlHaskellJavaJavaScriptJSONJSXLatexLispMarkdownHTMLPerlPHPPythonRReasonRubyScalaSQLSwiftVisual Basic

Each line in the text file, will be coming in as input (from the text file we gave in the first entry of the pipeline) and the output will be the absolute path to the GCS bucket.

The following code unzips each file, read it as a string, then hands over to the next one in the pipeline:

    @ProcessElement
    public void processElement(ProcessContext c){
        String p = c.element();
        GcsUtil.GcsUtilFactory factory = new GcsUtil.GcsUtilFactory();
        GcsUtil u = factory.create(c.getPipelineOptions());
        byte[] buffer = new byte[50*1024];
        ZipInputStream zis = null;
        ZipEntry ze = null;
        try{
            SeekableByteChannel sek = u.open(GcsPath.fromUri(p));
            InputStream is = Channels.newInputStream(sek);
            BufferedInputStream bis = new BufferedInputStream(is);
            zis = new ZipInputStream(bis);
            ze = zis.getNextEntry();
            while(ze!=null){
                StringBuilder zipContents = new StringBuilder();
                log.info("Unzipping File {}" + ze.getName());
                int len;
                while((len=zis.read(buffer))>0){
                    zipContents.append(new String(buffer, 0, len, "UTF-8"));
                }
                log.info("File " + p + " is opened successfully.");
                ze=zis.getNextEntry();
                c.output(zipContents.toString());
                TOTAL_JSON_FILES_COUNT.inc();
            }

        }
        catch(Exception e){
            log.error("An exception processing the zip file - " + c.element(), e);
        } finally {
            try {
                if (zis != null) {
                    zis.closeEntry();
                    zis.close();
                }
            } catch(Exception ex){
                log.error("Unable to close the zip files: " + c.element(), ex);
            }
        }
    }







Finally wire them all:





Pipeline pipeline = Pipeline.create(pipelineOptions);


        try {
            pipeline.apply("Get Zip File folders", TextIO.read().from("gs://rchandra/input/listofzipfiles.txt"))
                    .apply("Get GCS Path for each zip file", ParDo.of(new GetGCSUtilPath()))
                    .apply("Reading Zip Files",ParDo.of(new ProcessZipFolder()))
                    .apply("Process Json Files", ParDo.of(new ProcessJsonData()))
                    .apply("Writing results", TextIO.write().withSuffix(".csv").withNumShards(10).to("gs://output/"));

            pipeline.run();
        } catch(Exception ex){
            log.error("Exception while running the zip files");
        }









The above code reads the zip file, unzips the json files in them, process them and then finally writes the output as a csv file in the given output location.




A sample "listofzipfile.txt" can be like this:




gs://rchandra/input/bucket1/*.zip
gs://rchandra/input/bucket2/*.zip
gs://rchandra/input/bucket3/*.zip
gs://rchandra/input/bucket4/*.zip
gs://rchandra/input/bucket5/*.zip

Adding line numbers to a large number of entries in a file

After looooong time...

I had this requirement of creating a sequence file for a Map-Reduce job.  I had a large set of files (some 1000s) in a folder and I need to create keys.txt for the Sequence file creator.

First I thought I need a painful way of writing a python module to get the list of files, add the index value as I walk through the list and then save them in a file named keys.txt

Then I found out the good old editor vi can easily do this job for me.  Here are the steps:

1) Go to the folder where you have the last set of files
2) Use ls -1a > keys.txt to get just the file names.  One caveat is this will add ., .. and keys.txt in the file list.  We need to remove them.
3) Open the keys.txt in the vi.
4) Remove those 3 lines that has those entries mentioned in 2).
5) In vi command mode, enter the following:

%s/^/\=printf('%-3d,', line('.'))

and press enter.

The above command will add the line number for each line. For example:
1,1.zip
2,2.zip
...
...
...
500000,500000.zip

And that is all folks....

Calculating HBase Tables disk space

If you want to get the size of the HBase tables,  the simple way is to use 'hadoop' command from your local hadoop setup.  The command is:

hadoop --config <your hbase-config-file location> fs -du  -h <folder where the data resides>

For example:
hadoop --config c:\hadoop\staging fs -du  -h /hbase/data/default/

This will display all the tables under that 'default' folder (Note: Based on your setup, the data folder may vary, but you get the idea).  The first column is the actual table size across the region servers.  The second column shows the total size including the replication.  The second column is important in deciding how much "total" space is required.

1.4 M    4.3 M    /hbase/data/default/Table1
79.9 M   239.8 M  /hbase/data/default/Table2
2.0 K    6.0 K    /hbase/data/default/Table3

Cassandra:nodetool usage

Nodetool is a OOTB tool from Cassandra.  It provides various options to manage the keyspace.  There were three useful options I came across:

1) I had to copy the keyspace from one server to another server.  Since the source is from multiple servers, it is essential to cleanup the data after all the db and its related files are copied.  For that, I have to use nodetool.  Nodetool usually resides in <cassandra-path>/bin.  To cleanup, give the following command:

            nodetool -h <hostname of the cassandra> cleanup

Based on the amount of data to be cleaned, it will take some time.To track the progress, you can use:

            nodetool compactionstats

This will print the following:

pending tasks: 1
          compaction type        keyspace   column family       completed           total      unit  progress
                  Cleanup             <keyspace-name>      <column-family-name>      1667500496      2004577146     bytes    83.18%

Note: the total column will vary based on the size that is being calculated in real time.

2) To get the statistics of all the keyspaces. Especially to get a rough idea of how much keys per column family are being used:

           nodetool -h <hostname> cfstats

MSSQL: How to transform a set of rows from one table to another?

My requirement was to transfer a specific set of rows from one table to another table and both have the exact schema bit-by-bit.  In MSSQL, follow these steps:

1) First get the column list from the destination table.  Use:

SELECT SUBSTRING(
    (SELECT ', ' + QUOTENAME(COLUMN_NAME)
        FROM INFORMATION_SCHEMA.COLUMNS
        WHERE TABLE_NAME = '<destination-table-name>'
        ORDER BY ORDINAL_POSITION
        FOR XML path('')),
    3,
    200000);

2) If the destination table has the ID column (primary key), then you need to turn on the identity_insert option:
   
     set identity_insert <destination-table> on;

3) Then you insert...into...select query.  The select can have any valid condition to get the specific set of data.  A typical query can be like this:

insert into emails (<destination-table-column-list-from the query in step 1) SELECT <source-table-column-names>
  FROM <source-table> where <any condition that will load only those records>;

4) Don't forget to turn off the identity_insert option you changed in step 2):
 
    set identity_insert dbo.emails off;

And that is all...

Installing Jupyter Notebook and Numpy in Windows

I had hard time to install Jupyter Notebook.  My machine already had Python 2.7 and after installing anaconda and trying to run 'juypter notebook' command threw python errors.

Then I tried to uninstall Python 2.7 and again let the anaconda install its own python version.  No luck.

Then finally I removed all the Python installation but forgot to remove the path variables (more on this below).  Then installed python 3.5 version (take the latest 3.5 release).

Then using pip3 install jupyter, the notebook got installed successfully.

Then running 'jupyter notebook', got an error message "Fatal Python error: Py_Initialize: unable to load the file system codec".  Upon checking the forums, one of the solution was to check the PATH and the PYTHONPATH environment variables.  Bingo..they were still referring to the older and now removed installations of Python.

Removed them and the 'jupyter notebook' brought the index page in the browser but with the 404 error.

Again, forums helped me.  It looks like there may be a bug in the jupyter version I downloaded, so if you run the 'jupyter notebook' from your root folder (c:\), then this error will happen.  This could be due to the fact that the root folder has hidden system files.   Switch to any folder, say, c:\tmp, and run the command, brings the page with the file contents of that folder.

Numpy is bit different.  Since there is no installation manager provided by numpy.org, you need to first download the binary wheel for the python version you're using.  Important: Make sure you are downloading the right version of the whl file.  if you're using Python 3.5, look for the one with the name that contains cp35.

Then go to that downloaded folder and run pip install <whl file name>.  For example, pip3 install "numpy-1.11.2+mkl-cp35-cp35m-win32.whl".  This will install the numpy package.

That's all folks!.

Using Jboss JConsole

If you're using JBoss as your app server, then you don't need to setup JDK settings for the JMX.  Instead you need to check the following: http://stackoverflow.com/questions/17105933/enabling-jmx-remote-in-jboss-6-1

Then run JBOSS_HOME/bin/jconsole.sh (or .bat for Windows).

In the "Remote" section enter the following:service:jmx:remoting-jmx://<server-name>:9999 (the default OOTB port number is 9999.  Check 'management-native' value in standalone.xml.

Then enter the username and password we created using add-user.sh (or .bat) in jboss/bin.

Using Jstatd

JVisuslVM is useful in profiling your server applications local or remote.  To profile your application, you need to run the jstatd daemon in the server your application is running.  Running jstatd is not as straightforward.  Follow these steps:

1) Create a policy file.  The policy tool is in JAVA_HOME/jre/bin.  Start policytool.exe

2) You need to provide the security policy for the tools.jar.  The tools.jar is usually in JAVA_HOME/lib.  DO NOT GIVE RELATIVE PATH such as $JAVA_HOME/lib/tools.jar.  The jstatd doesn't recognize it.  Instead provide the absolute path.

3) Click "Add Policy Entry" and enter the following in "CodeBase" text box: file:/usr/java/default/lib/tools.jar (for non-windows and also, check your java path in your machine) or "c:/Program Files (x86)/Java/jdk1.7.0_79/lib/tools.jar" (for Windows, for ex.  Use "/" for windows path also).

4) Click "Add Permission" and enter the following in "AllPermission" text box: java.security.AllPermission. Click Ok.

5) Click "Done"

6) Click File->Save and save it.

7) Copy this policy file to the remote server (if needed).

8) Go to $JAVA_HOME/lib folder.

9) Run the following: jstatd -J-Djava.security.policy=/<absolute path to the folder where the policy file is>/<policyfile name> -p1234

Note: The -p option may not  required in some cases and this value is a random value chosen.

Now run the Java VisualVM and Click on Remote and add the remote server name.  You should see  the Jstatd process.

Switching SOLR JDK version

In some cases, your JAVA_HOME environment variable may not point to the  version SOLR demands.  For example, my development environment demands JDK 6, where as SOLR 5,x requires JDK 1.7 and above.

SOLR provides its own JAVA_HOME environment variable SOLR_JAVA_HOME.  Set this variable to the JDK version you need to and you are all set.

In Windows, use the Control Panel->Environment Variables section to set the SOLR_JAVA_HOME.

Then run solr start -p <port-number>, usually, <port-number> = 8983.

That is all.

Using serialver JDK command

If you are implementing java.io.Serializable, it is a good practice to create the serialVersionUID for that class.  Otherwise, when the class structure is changed (adding or removing the attributes), you will have the nightmare of handling the different serialVersionUID issues.

You can provide any random value to the serialVersionUID but it is better to let the tools generate the unique IDs for each class.

Eclipse comes with the built-in serialVersionUID option to generate the unique UID.  However, if the project is not compiled successfully (typically the dependencies issues), or if you don't want Eclipse to build your project, generating the version UID is not possible.

That is where serialver command line comes to the rescue.  The format is:

serialver [-classpath classpath] [-show] [className]

[-classpath classpath] - is where your class files' root folder.  For ex. if you have <myfolder>/target/classes/com/mytest/MyOnlyTest.class, then your class files root folder is <myfolder>/target/classes/.  

Most of the times the classpath value is required.  If you run the serialver command from "<myfolder>/target/classes", the classpath value will be "./".  It is always better to start from the folder where the root package is the subdirectory of that.

[-show] - This will provide UI option.  This will work just like the command line.

[className] - The fully qualified class name (that is including the package name).  For ex. com.mytest.MyOnlyTest

Example  (Using the command line):

c:\<myfolder>\target\classes>serialver -classpath "./" com.mytest.MyOnlyTest

If the class is a valid one, you will see the serialVersionUID entry like this:

com.mytest.MyOnlyTest:  static final long serialVersionUID = -7039383885342506716L

Copy that value and paste it in your java file.

Example (Using the UI):

c:\<myfolder>\target\classes>serialver -classpath "./" -show

This will display a small box with entering the full class name.  Enter com.mytest.MyOnlyTest, then click "Show".  If the class is a valid one, you will see the serialVersionUID entry in the "Serial Version" box.  

Copy that value and paste it in your java file.

Note: If you get the <className> is not found, check whether you entered the right class path value.

Reverting accidental SVN Checkins

The one thing I want from TortoiseSVN is a confirmation message whenever I am commiting my changes.  Sometimes, I use "SVN Commit..." option to see which files were changed and accidentally, I hit "Ok" instead of "Cancel".   If you're in the same boat, read on.

If you accidentally checked in the files that are not supposed to be, you need to revert that revision and bring your local code back to the same set of files that must be in the "Changed" state.  Follow these steps for the result:

1. Create a new local folder.
2. Get the latest code from the branch where the files are accidentally checked in.
3. Right click on the local folder and select "Tottoise SVN...Show Log". This will show the revisions including the one you had accidentally checked in.
4. Right click the revision below the one you last checked in.
5. Choose "Revert to this revision" option.
6. This will cancel all the changes that were checked in the "accidental" version in your local copy.
7. Right click on the new local folder and select "SVN Commit..." to commit the reverted changes.
8. To bring back your changes to the local folder, see the next section "Bringing back the changes to the local version".

Bringing back the changes to the local version

1. Go back to the root of the folder where the "accidental" check in happened.
2. Right click and "Tortoise SVN...Show Log".
3. Click the revision  you checked in accidentally and the revision below that.
4. Right Click and select "Show differences as unified Diff".  This will show all the changes you did.
5. Click "File" and save it as a patch.
6. In the root of the folder, right click and "Svn Update"
7. It will bring the latest version (which had been checked in after cancelling out your changes).
8. Right click on the root folder and select "Tortoise SVN...Apply Patch..." option.
9. Select the patch file you created in 5).
10. Select "Patch All Items" option.
11. Your changes should now be available in the local branch.

Factory Pattern using Spring injection

In this article, let us see how to use Spring injection to use the factory design pattern.

What is factory design pattern?

From wikipedia, the factory design pattern is:"In class-based programming, the factory method pattern is a creational pattern which uses factory methods to deal with the problem of creating objects without specifying the exact class of object that will be created. This is done by creating objects via calling a factory method—either specified in an interface and implemented by child classes, or implemented in a base class and optionally overridden by derived classes—rather than by calling a constructor."

The advantage of the factory pattern is the caller just sends the information about which class the caller would like to execute based on the runtime scenario.  A typical example is when you want to encrypt a string based on different RSA algorithms, you can write the encrytion methods in individual classes and at run time, the caller can choose to call the factory method to specify which algorithm he needs to encrypt the data.

To implement the above example, I have created an interface called ICrypto which has two methods: encrypt and decrypt:

           public interface ICrypto{
             public byte[] encrypt(byte[] plaintext);
             public byte[] decrypt(byte[] encryptedText);
           }

The implementation of the above interface for AES will be:

           public class AESCrypto implements ICrypto {
             public AESCrypto() {}
             public byte[] encrypt(byte[] plaintext){
               ...
               //your code here
               ...
               return encryptedBytes;
             }

             public byte[] decrypt(byte[] encryptedtext){
               ...
               //your code here
               ...
               return plainTextInBytes;
             }
           }

Similarly, for RC2, the implementation is:

           public class RC2Crypto implements ICrypto {
             public RC2Crypto() {}
             public byte[] encrypt(byte[] plaintext){
               ...
               //your code here
               ...
               return encryptedBytes;
             }

             public byte[] decrypt(byte[] encryptedtext){
               ...
               //your code here
               ...
               return plainTextInBytes;
             }
           }

Define the crypto types enum:

           public enum CryptoTypes{
             AES_CRYPTO_TYPE,
             RC2_CRYPTO_TYPE;
           }

Now we need a factory class which is going to create the instance of one of the implementations above:

           public final class CryptoFactory {
             //make sure the constructor is not accessible.
             private CryptoFactory(){}
             //factory method
             public final static ICrypto getCryptoInstance(CryptoTypes type) {
               switch(type){
                case AES_CRYPTO_TYPE:
                 return new AESCrypto();
                case RC2_CRYPTO_TYPE:
                 return RC2Crypto();
               }
             }
           }

Now the caller can call the CryptoFactory.getCryptoInstance() with the enum value as a parameter and get the instance,

The above is a typical example of the factory pattern in non-spring world.

In Spring, all of the above are the same except the factory class.  Using the injection mechanism, the factory class need not go through the switch() statement.  Lets see how this can be achieved:

1.  First define all the implemented beans in the spring bean definition XML.For example:
   <bean id="crypto.rc2crypto" class="com.example.RC2Crypto"></bean>
2. Then define the enum constants also as a bean instance.  For example:                               <bean id="ctyptotype.enum.rc2" class="com.example.CryptoTypes" factory-method="valueOf"            <constructor-arg value="RC2_CRYPTO_TYPE"/>                                                                 </bean>
3. Now wire up the crypto type bean definitions with the implementation bean definitions as a map.  For example:                                                                                                                                    <util:map id="cryptoMap" map-class="java.util.LinkedHashMap">
   <entry key-ref="ctyptotype.enum.rc2" value-ref="crypto.rc2crypto" />
   <entry key-ref="ctyptotype.enum.aes" value-ref="crypto.aescrypto" />
      </util:map>
4. Wire the factory bean instance with the map we created in 3.  For example:                    <bean id="crypto.factory" class="com.example.CryptoFactory">                                                            <property name="cryptoTypes" ref="cryptoMap" />                                                   </bean>

With the above implementation, the same CryptoFactory will look like this:

           public final class CryptoFactory {
             private Map cryptoTypes;

             public Map getCryptoTypes() {
               return cryptoTypes;
             }

             public void setcryptoTypes(Map cryptoTypes) {
               this.cryptoTypes = cryptoTypes;
             }

             public ICrypto createCryptoFactory(CryptoTypes cryptoType){
               return cryptoTypes.get(cryptoType);
             }
           }

Now the caller can inject the instance of the crypto factory bean in their bean class and call createCryptoFactory(CryptoType.AES_CRYPTO_TYPE).

Even better way is the caller bean can define allowed crypto types for that bean (for example, the bean can use only AES_CRYPTO_TYPE) and use that variable instance to call createCryptoFactory().   It is now boiled down to the bean implementor's design.

Advantages of using Spring injection

As I mentioned above, once we defined all the possible types as enum, the caller bean can control which subset of enum types it can use to get the crypto implmentation's instance.  This adds more security as module X in the application can use only AES crypto type whereas module Y can use only RC2 crypto type.

Also, turning on or off of any crypto type is going to be only in the spring context XML files rather than in the code.  The factory class is encapsulated from which crypto types it is using.

And that is all folks,,,,

Useful Regular Expressions

Often times we need to search and if needed, replace the text in the large document or extract logs based on some fixed string values.  This post lists some of the regular expressions I have used very often.  This post will be updated regularly as I come across new regular expressions for new use cases:

Use Case 1:

Find a position if the first occurrence of the string.  for example, find the first occurrence of the string "SF_".  This will highlight the line up to that string.

^.*?(?=SF_)

Use Case 2:

Select whole line.

(.+)

Use Case 3:

Surround the entire line with the double quotes and add "," after the endquote.

"\1",

Use Case 4:

Find end of the line and add "," at the end.

Search: ([^\n])$
Replace : \1,

Use Case 5:

Find a string and highlight up to the end of the line.

Search: <string>.*

Example:
query.*


That is all for now.  As I mentioned in the beginning, I will update this post as I come across interesting scenarios.

JSON Schema validator

Like XML, JSON also provides schema and the code has to validate the json instance with a schema.  In this post, let us see how we can validate the given instance with its JSON schema.

The validator I would recommend is 'json-schema-validator'.  The version I have used is 2.0.1, which  supports both v3 and v4 Json schemas.  You can take a look at the latest versions of the validator also.

Add the following in your pom.xml:

<dependency>

    <groupId>com.github.fge</groupId>
    <artifactId>json-schema-validator</artifactId>
    <version>2.0.1</version>
</dependency>

Then follow these steps:

1) If the schema or the instance are strings, transform them to JsonNode objects:

            ObjectMapper mapper = new ObjectMapper();
             //json instance object
            JsonNode instanceJsonObject = mapper.readTree(jsonInstance);
            //json schema string to json object.
            JsonNode schemaJsonObject = mapper.readTree(jsonSchema);

2) Next step is to get the json schema object:

             //The schema factory supports both Json's v3 and v4 schemas.  Most of the times, the default
             //factory is enough.
             JsonSchemaFactory schemaFactory = JsonSchemaFactory.byDefault();            
             JsonSchema schema = schemaFactory.getJsonSchema(schemaJsonObject );

3) Now the instance can be validated with the validate() of JsonSchema:

             ListProcessingReport report = (ListProcessingReport)schema.validate(instanceNode);

The ListProcessing contains the valiation messages, the missing required field(s) details.  To obtain those details, follow these steps:

            if(!report.isSuccess()){

                JsonNode reportRootNode = report.asJson();            

                if(rootNode.size() > 0){
                    JsonNode errorNode = rootNode.get(0);
                    JsonNode messageNode = errorNode.get("message");
                    if(messageNode != null){
                        logger.error(messageNode.asText());
                    }
                    JsonNode requiredNode = errorNode.get("required");
                    if(requiredNode != null) {
                        logger.error(requiredNode.toString());
                    }
                    JsonNode missingNode = errorNode.get("missing");
                    if(missingNode != null){
                        logger.error((missingNode.toString());
                    }

                }
            }

And...that's all folks.