Wuzzuf, is a technology firm founded in 2009 and one of the very few companies in the MENA region specialized in developing Innovative Online Recruitment Solutions for top enterprises and organizations, They successfully served 10,000+ top companies and employers in Egypt, 1.5 MILLION CVs were viewed on their platform and 100,000+ job seekers directly hired through them. In total, 250,000+ open job vacancies were advertised and now, 500,000+ users visit their website each month looking for jobs at top Employers.

Wuzzuf, has released a sample dataset on Kaggle (Which provides data science competitions, Datasets, and Kernels), named Wuzzuf Job Posts. The dataset contains 2 CSV files:

• Wuzzuf_Job_Posts_Sample.csv: which contains Wuzzuf job posts with following attributes:

  • id: post identifier 

  • city_name: is the city of the job.

  • job_title: the title of the job

  • job_category_1, job_category_2 and job_category_3: which contains the most 3 relevant categories of the job post, e.g., Sales/Retail/Business Development

  • job_industry_1, job_industry_2 and job_industry_3:  which contains the most 3 relevant industries of the job post, e.g., Telecommunications Services

  • salary_minimum and salary_maximum: the salary limits.

  • num_vacancies: how many open vacancies for this job post.

  • career_level: enumeration of career levels e.g., Experienced (Non-Manager) and Entry Level

  • experience_years: number of years of experience.

  • post_date: publication timestamp of the post. e.g., 2014-01-02 16:01:26

  • views: count of views

  • job_description: detailed description for the job post.

  • job_requirements: main job requirements for the job post.

  • payment_period: salary payment interval e.g. Per Month

  • currency: salary currency e.g. Egyptian Pound

• Wuzzuf_Applications_Sample.csv.zip: Which contains Wuzzuf job applications, it have the following attributes:

  • id : application identifier

  • user_id: applicant identifier

  • job_id: post identifier

  • app_date: application timestamp, e.g., 2014-01-01 07:27:52

Data Cleaning

The published data-set had many free-text fields, Wuzzuf system does not enforce a certain list of items to choose for them, which makes processing and aggregation difficult. A common handling such as lower case all values and remove trailing spaces was performed. Additionally some fields needed special handling such as:

1. city_name:

   • this attribute is free text attribute, which represents Egyptian cities, but it has the following problems:

     • Misspelling of words. (i.e. cairo , ciro , ciaro).

     • Arabic names (i.e. القاهرة )

     • Outside Egypt cities (i.e. riyadh, doha)

     • General Cities (i.e. all egypt cities , any location)

     • Group of Cities (i.e."cairo, alexandria - damanhor")

   • All the above issues has been solved by: 

     • Outside Egypt cities: a static list of outside cities has been mapped to category "outside".

     • Arabic (Non-ascii) names: has been replaced statically be the corresponding english words.

     • General Cities: a static list of outside cities has been mapped to category "any"

     • Remove Not Needed substrings such as "el" and "al".

     • Replace "and" and "or" substrings with "-" to be splitted on next steps.

     • Group of Cities : attribute has been splitted on several delimiters.

     • Misspelling of words: a static list of valid cities and its states in Egypt has been created , each misspelled word has been mapped to the most similar word of valid cities, a threshold T has been used to accept only similarities above that threshold, otherwise city will mapped to "any" category.

     • Added new state attribute by mapping each city_name to its state from valid cities & states categories.

2. job_category_1, job_category_2, job_category_3 attributes: cleaning was done by removing placeholder text "Select" from all 3 attributes, and merging the 3 attributes into one attribute called job_categoriesd

3. job_industry_1, job_industry_2, job_industry_3 attributes: cleaning was done by removing placeholder text "Select" from all 3 attributes, and merging the 3 attributes into one attribute called job_industries.

4. experience_years attribute: we manually normalizing free text onto one of 3 forms 'x+' or 'x-y' or 'x', and split the experience_years attribute to 2 new attributes experience_years_min and experience_years_max , which contains the minimum and maximum years respectively needed for a job.

5. post_date attribute: we generated a new attribute called "post_timestamp" which has the POSIX timestamp value of the post_date attribute (i.e., the number of seconds that have elapsed since January 1, 1970 midnight UTC/GMT)

6. job_description and job_requirements attributes: we noticed that job_requirements attribute are normally empty, so we added new derived attribute called "description" which contains the concatenation of job_requirements and job_description attributes.

Derived Attributes

The next step was deriving some attributes from these data sets. We derived the following attributes:

1. Tags attributes: we used a third party API from MeaningCloud to extract Tags from the "description" attribute (recall that it contains the data from job_description and job_requirements). Thus, we added to the data-set the following attributes: 

   • quotation_list: which represents quoted text. e.g., you take on the responsibility of growing the Academy by increasing business and handling operational and technical challenges that arise in the process.

   • entity_list : which represents named entities as people, organization, places, etc. e.g. MS Office, Word, Excel, Weeks and Cairo

   • concept_list: which represents significant keywords. e.g., ability, system, software, code and computer science.

   • relation_list: This attribute could be used to provide a summary for the description attribute as it highlights most of the important notes from the description part.

   • money_expression_list: which represents money expressions, e.g., 2000 EGP

   • time_expression_list: which represents time expressions, e.g., 6 Months at least and 8.5 hours

   • other_expression_list: which contains other expressions such as alphanumeric patterns. e.g., php5

2. applications_count attribute to each post, which calculates how many applicants has been applied to this job post. (derived from applications data-set)

3. first_applicant_timestamp and  last_applicant_timestamp attributes per each post, which calculates the POSIX timestamp of the first and last applicant that applied to this job post.  (derived from applications data-set)

Case Study: Jobs Recommendation

Extracting tags from the jobs opens the doors for recommending jobs for applicants. We exploit the entity_list and concept_list attributes to rank the job posts that are relevant to the given applicant. On the other hand, we build a keywords vector from the applicant profile. The recommendation selection works calculating the 10 highest matching scores using a heuristic model. 

As a proof of concept, we analyzed some applicants profiles (their private information such as names was removed for anonymity). The following is a sample of the analyzed profiles:

Our system recommends the following job posts to him (ordered from best or lowest):

• System Administrator, with job_id = "8c872132", with score = 1.0

• System Administrator, with job_id = "a13539c", with score = 1.0

• System Administrator, with job_id = "c820bb65", with score = 1.0

• Technical Support Engineer French Speaker, with job_id = "6783a66f", with score = 1.0

• Data Entry & IT Technician, with job_id = "8c872132", with score = 1.0

• Software Developer SharePoint, with job_id = "990d3300", with score = 1.0

• .Net Developer, with job_id = "22a298c7", with score = 1.0

• Operations Support Engineer, with job_id = "69318c48", with score = 1.0

• Microsoft Product Manager, with job_id = "eb59b18d", with score = 0.8571

For further details about these job posts, check the dataset using these IDs.

Case Study: Job Summary

In some use cases, it is useful to summarize a bulk of text and get the most relevant information from a given text. As mentioned before, we added the relation_list attribute which highlights most of the important notes from the description part. Using this attribute, we can provide a short, yet descriptive, summary of the post. As an example, here is the original job post description and its summary for job post number 68417a3c.
 

The original description contains around 1317 characters while the summarized one contains only 544 characters, which an approx 59% reduction in size.

It's our pleasure to highligh the initiative taken by our data team leader Ahmed Mahran to effectively contribute to the Spark Time Series project, created by Sandy Ryza, a senior data scientist at Cloudera, the leading big data solutions provider.

Time Series data has gained an increasing attention in the past few years. To quote Sandy Ryza:

Time-series analysis is becoming mainstream across multiple data-rich industries. The new Spark-TS library helps analysts and data scientists focus on business questions, not on building their own algorithms.

Find the full story here, where he introduces SparkTS, and accredits our contributor.

We are, forever, indebted to the open source community, it enabled us to create wonderful feats. It's our deep belief that we should give back to the community in order to guarantee its health and sustainability. We are proud that we effectively contributed to such great project and we are looking forward to more.

   
    In search space, pagination always has to happen. Solr has the feature of basic paging. In basic paging, you simply specify start and rows parameters. start indicates where the returned results should start and rows specifies how many documents are returned. With basic paging, partial index exporting and migration is a problem. Since basic paging needs to sort all the results first before returning the desired subset, it needs large amount of memory if start is of high order. For instance, start=1000000 and rows=10 causes an inefficient memory allocation to happen due to the sorting of 1000010 documents. In a distributed environment, the case is worse because the engine has to fetch 1 million documents from each shard, sort them then return the result set.

Shared storage and HDFS (pros and cons)

Figure 1 - Shared storage in a distributed compute cluster.

HDFS was built to scale out using disks directly attached to commodity servers so as to form a basis for a distributed processing framework that exploits the data locality principle - moving code to data is much more cheaper than the other way around. Hence, incorporating a shared storage system, as opposed to directly attached storage (DAS), in order to host the HDFS is not straightforward as centralized storage is not the nature of the HDFS.

Moreover, shared storage systems have some potential limitations. As a shared service, it needs to efficiently handle multiple concurrent IO requests from the numerous clients of the compute cluster. Also, the network is the channel for communicating data from and to the shared storage, the thing that would consume much of the network’s bandwidth. While in a DAS model the network is the communication channel as well for sharing data, data local computations reduce network bandwidth consumption significantly. Moreover, comparing network bandwidth to the regular internal per-machine data busses, data busses are faster as, besides the underlying hardware technology, they don’t incur the overhead imposed by the networking stack which constitutes the network infrastructure and protocols (layered protocols headers, checksum computations …).

On the other hand, shared storage systems have unique attractive features that worth to be considered either while constructing a new distributed data processing cluster or when there is a need to exploit an existing infrastructure with a pre-installed shared storage system. One important feature is separating compute from storage resources. This implies independent administration and scalability of resources; it would be possible to scale only one resource type, either compute or store, without unnecessarily scaling the other. In addition to that, compute resources will be almost fully utilized in computations rather than taking the overhead of servicing shared data to other consumers. The thing that would ensure efficiency and high availability for data sharing.

Moreover, as a plus, shared storage systems provide application agnostic features releasing more overhead from compute resources while adding more strengths to the cluster. Those features like: security, backups and snapshots, efficient capacity utilization via compression, enhanced IO performance by leveraging the latest storage and network technologies, and failure handling and high availability. So, with shared storage system, the namenode is no more a single point of failure as the metadata could be backed up safely without the need for other namenodes designed as part of the Hadoop HA (High Availability) feature. Besides, data replication, as an inherent feature in shared storage systems, relieves block replication burden off HDFS daemons saving CPU, memory and network bandwidth as lesser block replications simplifies write pipelines and downsizes block pools.

How HDFS works with shared storage

For HDFS, we can categories shared storage systems into two main groups: HDFS aware and HDFS agnostic. HDFS aware storage systems are those that provide native support for HDFS sometimes to the extent that HDFS daemons are running inside the box of the storage system. In that case, HDFS is readily tuned for the shared storage system, moreover, the rest of the cluster resources become pure compute resources. Alternatively, HDFS agnostic storage systems don’t know about the existence of the HDFS hence their incorporation has much more involvement.

For an HDFS agnostic shared storage system, aside from the recommendations of the storage system provider, a typical deployment would be to allocate separate storage partitions for each host in the cluster and to distribute HDFS daemons such that one or two master hosts to have the namenode and the secondary namenode daemons and each of the rest of the slave hosts to have at least one datanode daemon. Given such configurations, let’s inspect the cost per read and write operations.

Figure 2 - Data flow while Host 2 performs a write with replication factor of 2.

Figure 2 illustrates the data flow if a client on Host 2 is writing two replicas, one local on Host 2 and another remote on Host 1. The following are the paths for the data transmission:

1. The client on Host 2 transmits the data to the colocal data node over kernel network.
2. The data node on Host 2 transmits the data to the shared storage over network.
3. The data node on Host 2 transmits the data to the data node on Host 1 over network.
4. The data node on Host 1 transmits the data to the shared storage over network.

The same data is transmitted over network three times and is written on the same storage twice. Hence, write operations are very expensive in terms of network bandwidth and storage capacity consumption. It is meaningless to consider block replication on a shared storage especially if the shared storage system readily provide data replication.

Figure 3 - Data flow while either hosts performs a read.

In Figure 3, if a client on Host 1 is reading a block maintained by the colocal data node, the following are the possible data transmission paths:

1. The data is transmitted over network from the shared storage to the data node on Host 1 (link C).
2. The data node on Host 1 transmits the data to the colocal client (link A) over kernel network.

However, if a client on Host 2 reads a block only maintained by the remote data node on Host 1, possible paths are:

1. The data is transmitted over network from the shared storage to the data node on Host 1 (link C).
2. The data node on Host 1 transmits the data over network to the remote client on Host 2 (link B).

In a remote read, the same data is transmitted over network twice while only once in local reads. Hence, remote reads are expensive than local reads in terms of network bandwidth consumption. The following table summarizes the cost of data transmission in the different scenarios discussed above.

Increasing block replication, increases data locality and read performance however decreases the write performance and wastes storage. It would be great if there is a way to minimize the length of data paths by taking shortcuts.

Short-circuit reads

Figure 4 - Short-circuiting local and remote reads from a shared storage.

Short-circuit local reads

HDFS provides one shortcut for local reads. If the client is reading a file from a collocated datanode, the client can read the file directly from the disk bypassing the datanode. In figure 4, a client on Host 1 requests to read a file from the colocal datanode. Then, the datanode redirects the client to the required block in the filesystem and the client reads the file directly from the shared storage over network.

Short-circuit remote reads

By analogy from the short-circuit local read, a short-circuit remote read is defined when a client requests a read from a remote datanode, then the remote datanode provides direct access to the file for the client. This could be a valid scenario if only the data is stored on a shared storage system to which all hosts have direct access. However, HDFS is not aware whether it’s hosted on a shared storage or not. Hence, HDFS is not providing a shortcut for remote reads but it is still possible to hack around it.

HDFS client knows whether it’s colocal with a datanode by checking the datanode IP address against the list of network interfaces on the same host as the client. If the datanode IP is one of them, the client knows that the datanode is colocal. Then, there are two possible ways for the client to get the file: the legacy short-circuit read, or the new short-circuit read. In the legacy implementation, the client requests, through TCP, the absolute path of the block from the datanode. Then, the client opens the block file and reads the content directly after receiving the path info from the datanode. While in the new implementation, the client connects to the datanode through a Unix domain socket. Then, the datanode opens the file and bypasses the file descriptor via the Unix domain socket to the client. After that, the client can consume the file directly through the descriptor.

HDFS short-circuit read mainly involves the following three actions:

1. Datanode locality check
2. Interprocess communication through Unix domain socket (new implementation only)
3. File direct access

Datanode locality check: It is possible to come around this check by adding, to each host, a dummy network interface, one for each remote datanode. Each dummy interface is assigned the same IP address as the corresponding remote datanode. It’s important to hide these interfaces from network so as to be able to reach the real datanodes.

Interprocess communication through Unix domain socket: A Unix domain socket works on the kernel level and connects parties on the same host only. Processes meet at local files but share data through memory. To the extent of our knowledge, it’s possible to forward a Unix domain socket between two hosts hence enabling two remote processes to communicate through the Unix domain socket. For example, OpenSSH can do Unix domain socket forwarding. Hadoop provides only one configuration parameter for setting the socket address (dfs.domain.socket.path) such that the datanode and the DFS client on the same node use the same socket address which makes it difficult to forward the DFS client connection to the appropriate datanode. So, it’s not a straightforward task to hack the short-circuit new implementation for short-circuiting remote reads.

File direct access: Clients can have direct access to files maintained by a remote datanode if the files are hosted in a shared file system to which all hosts have access. Each datanode should be assigned an independent data directory that is hosted on a shared file system and that have the same mount point on all hosts. For example, datanode 1 on Host 1 is assigned a data directory /hdfs/data1/ and datanode 2 on Host 2 is assigned a data directory /hdfs/data2/. Both directories are hosted by a shared file system (say NFS) and are mounted on each host at the mount points /hdfs/data1/ and /hdfs/data2/, respectively.

Fortunately, we can hack around the legacy implementation of short-circuit local reads as it does not involve any Unix domain communication. The following table summarizes the cost of data transmission in the different read scenarios after enabling short-circuit remote reads. Unfortunately, HDFS does not provide a shortcut for writes hence the write cost stays the same.

Now, local and remote reads have the same reduced cost. This is a substantial improvement to HDFS for reading from a shared storage.

Proof of concept

Experiment Setup

Figure 5 - Short-circuit remote read experiment setup (SCRR is a shorthand for Short Circuit Remote Read).

As illustrated in figure 5, we build a small cluster of three machines with the following configurations. SCRR3 is the master node hosting an NFS server simulating the shared storage and exposing two locations (/home/host/hdfs1 and /home/host/hdfs2) as the storage directories for the cluster datanodes. It also hosts the cluster namenode and secondary namenode. SCRR2 and SCRR1 are both slave nodes each one hosting a datanode. The datanode on SCRR1 is configured with a storage directory that is hosted on the shared storage and mounted on /home/host/hdfs1. Also, the datanode on SCRR2 has a storage directory that is hosted on the shared storage however mounted on /home/host/hdfs2. Both storage directories are mounted on both nodes (SCRR1 and SCRR2) on the same mount points such that a datanode directory could be accessed on both nodes through the same path.

In our experiment, we assign each node a specific role. SCRR3 has the shared storage role, SCRR2 has the DFS client role and SCRR1 has the datanode role. We configure the HDFS cluster with minimum replication factor of one and we enable the legacy short circuit local read. We shutdown the datanode on SCRR2 and configure a dummy network interface with the same address of the node on SCRR1. We upload a file on the HDFS cluster (that would be kept by the datanode on SCRR1). We then run a simple DFS client on the node SCRR2 that reads the file byte by byte counting the number of bytes read. After the DFS client finishes we disable the short circuit remote read configuration by deleting the dummy network interface and then we re-run the same DFS client. We repeatedly run the same DFS client alternating between the two modes (short circuit enabled and short circuit disabled) while watching the network bandwidth consumption.

Results

Figure 6 - Network bandwidth consumption on each node.

Figure 6 shows the network bandwidth consumption on each node while performing the experiment. The short circuit is enabled in runs (B, D and F) and is disabled in runs (A, C, E and G). The following table summarizes the network bandwidth consumption on each node in terms of data transmission and reception rates as well as what data is being communicated. It is clear enough how short circuiting remote reads from shared storage reduces network bandwidth consumption significantly, not to mention the applications running time.

Appendix: Dummy network interface configuration on Linux:

Assume a cluster of three slaves

1. On each host, create a dummy network interface and assign it a suitable name, one for each remote datanode. For example, on Host1:

   sudo ip link add dummyDataNode2 type dummy
   sudo ip link add dummyDataNode3 type dummy

2. On each host, assign the same address of the real datanode for each device. For example, on Host 3:

   sudo ifconfig dummyDataNode1 1.1.1.1
   sudo ifconfig dummyDataNode2 1.1.1.2

3. Make sure that all machines have the MAC address of the real datanode. For example, on Host 1, assuming eth0 is the interface through which Host 1 is supposed to communicate with HDFS cluster network:

   sudo arp -s 1.1.1.2 00:00:00:00:00:02 -i eth0
   sudo arp -s 1.1.1.3 00:00:00:00:00:03 -i eth0

4. Make sure that all machines have a route to the real datanodes through the correct interface. For example, on Host 2:

   sudo route add -host 1.1.1.1 dev eth0
   sudo route add -host 1.1.1.3 dev eth0

5. On each host, hide the dummy interface from the network by disabling ARP for the dummy device and deleting all routing entries that involve the dummy device. Use both the route and ip-route commands to query for the routing entries involving the dummy device. For example, on Host 1:

	sudo ifconfig dummyDataNode2 -arp
	sudo ifconfig dummyDataNode3 -arp

	route

	Output:
	Kernel IP routing table
	Destination  Gateway    Genmask         Flags Metric Ref    Use Iface
	default      1.1.1.254  0.0.0.0         UG    0      0        0 eth0
	1.1.1.0      *          255.255.255.0   U     1      0        0 eth0
	1.1.1.0      *          255.255.255.0   U     1      0        0 dummyDataNode2
	1.1.1.0      *          255.255.255.0   U     1      0        0 dummyDataNode3

	sudo route del -net 1.1.1.0 netmask 255.255.255.0 dev dummyDataNode2
	sudo route del -net 1.1.1.0 netmask 255.255.255.0 dev dummyDataNode3

	ip route show table local dev dummyDataNode2

	Output:
	broadcast 1.1.1.0  proto kernel  scope link  src 1.1.1.2 
	local 1.1.1.2  proto kernel  scope host  src 1.1.1.2
	broadcast 1.1.1.255  proto kernel  scope link  src 1.1.1.2

	ip route show table local dev dummyDataNode3

	Output:
	broadcast 1.1.1.0  proto kernel  scope link  src 1.1.1.3
	local 1.1.1.3  proto kernel  scope host  src 1.1.1.3
	broadcast 1.1.1.255  proto kernel  scope link  src 1.1.1.3

	sudo ip route del table local dev dummyDataNode2 to local 1.1.1.2
	sudo ip route del table local dev dummyDataNode2 to broadcast 1.1.1.0
	sudo ip route del table local dev dummyDataNode2 to broadcast 1.1.1.255
	sudo ip route del table local dev dummyDataNode3 to local 1.1.1.3
	sudo ip route del table local dev dummyDataNode3 to broadcast 1.1.1.0
	sudo ip route del table local dev dummyDataNode3 to broadcast 1.1.1.255