Darius Kharazi's Blog

An Open-Source ML Pipeline

Sat, 03 Sep 2022 00:00:00 GMT

Whether our model is predicting churn, detecting fraud, or forecasting sales, there are a few components that are common across any ML pipeline. In particular, every ML pipeline includes the following features:

A feature store for model training and inference
Many CI/CD validations to detect changes in those features
An experimentation environment for logging previous model runs and metrics
A model registry for serving and version controlling production models
A service for tuning hyper-parameters efficiently
A service for model monitoring for automatic model retraining, data drift detection, and reporting

Using Feast as a Feature Store

A feature store is an API best used for low-latency feature retrieval for real-time models being served in production. A feature store plugs into your existing storage infrastructure and orchestrates jobs using your existing processing infrastructure, so it is not a database in most cases. Usually, a feature store takes already transformed features (from Azure Data Lake, AWS Redshift, Snowflake, etc.) and generates feature definitions and metadata for them, in order to improve the performance of the feature retrievals.

A feature store is used for maintaining features across different data sources in a single, centralized location. Doing this promotes a central catalog for all features, their definitions, and their metadata, which allows data scientists to search, discover, and collaborate on new features. Two common and open-source feature stores are Feast and Tecton. For more detailed use-cases surrounding feature stores, refer to Feast's documentation and Tecton's documentation, which are two of the most common and open-source feature stores.

In Feast's feature store, a feature store comes with two components: a registry and feature stores. The default Feast registry is a file-based registry, where feature definitions, metadata, and versions are tracked in your local file system under a file named registry.db. In production, Feast recommends using a more scalable SQL-based registry that is backed by a database, such as PostgreSQL or MySQL. The registry more specifically holds feature views. A feature view is an object that represents a logical, unmaterialized group of features consisting of the following information:

The data source
The specified features
A name to identify this feature view in Feast
Any additional metadata, like schema, description, or tags

The second component within Feast is the feature store, which can be an offline store or an online store. The offline store persists batch data from feature views. By default, the offline store will not log features and will instead run queries against the source data. However, offline stores can be configured to support feature writes to an offline destination. An online store is a database that stores only the last values for real-time inference. The online store is populated through materialization jobs from an offline store.

The offline store is preferred for fetching features when training a model or making daily or weekly predictions. On the other hand, the online store is preferred for fetching features when making real-time predictions (e.g. fraud detection). For more details about the feature registry, refer to their registry documentation and documentation about feature views. For more details about the various feature retrieval patterns, refer to their documentation.

Using Great Expectations for CI/CD Data Validations

Great Expectations is an API used for testing, profiling, and documenting expected feature properties, including expected value ranges and data types. The data profiler will automatically generate its own expectations in a report, illustrating example values, percentages of missing values for columns, histograms of numeric columns, etc. Manual tests can be included as well, which could include allowed column values, thresholds of null percentages before sendings warnings, etc. Great Expectations can be scheduled and orchestrated in a CI/CD pipeline using Airflow or Kubeflow.

By default, Feast uses Great Expectations as a validation engine and data profiler. As a result, we can specify the expected data types and value ranges for input columns. For a detailed example of using Great Expectations with Feast, refer to their documentation.

Using MLflow for Model Registry and Experiment Tracking

A model registry is a centralized object store that stores binary model objects with any metadata. By default, the model registry typically is stored in the local file system or a SQL database, but a remote object store also can be specified (e.g. AWS S3, ADLS, GCS, etc.). The metadata that is stored with the models could include model versions, stages, registry dates, and tags.

MLflow is one example of a model registry that documents a model's lifecycle. For example, MLflow will document the model versions and allow for the model to have different stages. Registered models can be in the production stage, staging stage, or archived stage. The production stage is meant for production-ready models serving inference, whereas the staging stage is meant for a pre-production model that is meant for testing and intended to be put into production in the future. Archived models are meant for previous models in production. For more details about the model registry component in MLflow, refer to the documentation.

Mlflow registers models that have gone through experimentation. A model experiment tracks and logs previous model runs with any specified metrics, hyper-parameter choices, training date, and other metadata. By doing this, we can compare previous model runs with each other by observing their feature importances, accuracies, hyper-parameters, etc. A tracking UI comes with MLflow, which allows you to visualize these model runs and download model artifacts or metadata. Refer to the MLflow documentation for more details about logging experiments in Mlflow.

Using Hyperopt for Distributed Hyper-Parameter Tuning

When logging model runs in an experiment, hyper-parameter searches can be logged using GridSearch or other packages like Hyperopt. Hyperopt is an API that facilitates distributed hyperparameter tuning and model selection. Hyperopt allows models to scan a set of hyperparameters across a specified or learned space.

The basic steps when using Hyperopt are the following:

Define an objective function to minimize
Define the hyperparameter search space
Specify the search algorithm
Run the Hyperopt function fmin()

In most cases, the objective function is the training or validation loss function. Hyperopt uses stochastic tuning algorithms that perform a more efficient search of hyperparameter space than a deterministic grid search. The fmin function executes a run by identifying the set of hyperparameters that minimizes the objective function. The fmin function accepts the objective function, hyper-parameter space, and an optional SparkTrials object. The SparkTrials object allows you to distribute each HyperOpt iteration from a single-machine tuning to the other Spark workers. For more high-level use-cases about Hyperopt, refer to the documentation in Databricks. For more details about the components of Hyperopt, refer to the documentation in Databricks.

The objective function is where we implement the training portion of the code, where a dictionary is returned with the desired loss function and status. For more details about SparkTrials, refer to the Databricks documentation. For more examples about defining a function for optimization, refer to the Hyperopt documentation. Note, if you're experiencing issues with memory, you may benefit from loading in the training/validation data as binary objects instead, since Hyperopt can be memory intensive. For a sample objective function, refer to the following code snippet:

# Initialize possible hyper-parameter space for tuning
space = {
    'learning_rate': hp.choice('learning_rate', [0.01, 0.02, 0.06, 0.08, 0.1]),
    'max_depth': hp.choice('max_depth', [3, 4, 5, 6, 7, 8]),
    'objective': 'reg:squarederror',
    'n_jobs': 10,
    'random_state': 3
    }

# Initialize objective function iterating over hyper-parameter space
def objective(space):
     
    # Autolog model details and signatures for each child run
    mlflow.xgboost.autolog()
    with mlflow.start_run(run_name=run_name, tags=tags, nested=True):
         
        # Load in training and test data from DBFS saved above to preserve memory
        train = xgb.DMatrix('/dbfs/FileStore/shared_uploads/darius_kharazi@anfcorp.com/train.buffer')
        validation = xgb.DMatrix('/dbfs/FileStore/shared_uploads/darius_kharazi@anfcorp.com/validation.buffer')
         
        # Initialize and fit model with different combinations of tuned hyper-parameters
        model = xgb.train(params=space, dtrain=train, evals=[(validation, "validation")])
 
        # Fit model and predict on test data
        test_y_ltv_pred = model.predict(validation)
 
        # Calculate accuracy metrics
        r = round(mean_squared_error(validation.get_label(), test_y_ltv_pred, squared=False), 3)
        m = round(mean_absolute_error(validation.get_label(), test_y_ltv_pred), 3)
 
        # Log RMSE and MAE metrics to mlflow experiment
        mlflow.log_metric('rmse', r)
        mlflow.log_metric('mae', m)
 
    # Return RMSE for hyperopt optimization
    return {'loss': -r, 'status': STATUS_OK}

When defining hyper-parameter ranges within a search space, use helper functions implemented in Hyperopt. Memory issues may appear if the ranges for particular hyper-parameters are too large. The SparkTrials objects is available to make use of distributed hyper-parameter tuning. For additional tips about defining a search space, refer to the Hyperopt documentation.

Using Evidently for Model Monitoring

Model monitoring involves maintaining registered or production-ready models and their metrics, data values, and other properties. Model maintenance can often be costly, since it impacts every data scientist and requires manual analysis about when and why accuracies worsen over time. The quality of a model in production can decline over time for many different reasons. For example, inventories could change over time, customer preferences could change over time, and regional changes could happen. Automating the model monitoring process offers the following benefits:

Automates the model retraining process
Automates metric and accuracy analysis

Evidently is an API that detects and alerts when there are changes in a model's quality. Evidently ensure models are retrained after a model drops below some accuracy threshold, which is manually assigned by the user. Evidently also saves and tracks any specified metrics and statistics for each model, which will be illustrated through visual reports.

Statistical tests to compare the input feature distributions can be triggered automatically, while alerts can be emailed alongside visual reports illustrating any drifts in the input features. By doing this, we can understand why model accuracies have changed over time based on drifts in the inputs. This highlights a common problem within model monitoring known as data drift, which specifically refers to distributional changes of any input features that a model is trained on. On a similar note, target drift may also occur, which refers to distributional changes of any model outputs (or predictions). Target drifts are caused by changes in the model inputs in most cases. For high-level details about model monitoring, refer to this webinar illustrating the occurrence of model staleness in production. For a detailed walkthrough of setting up and using Evidently for model monitoring, refer to this example in their documentation.

Goodhart's Law

Tue, 23 Aug 2022 00:00:00 GMT

I've recently become less reliant on measures like Yelp and Google reviews when selecting a restaurant. I've noticed many restaurants with many high Yelp reviews are sometimes not as good as restaurants with fewer Yelp reviews. On the flip side, many restaurants with lower ratings are sometimes better than those with higher ratings. To be clear, I stiil think these ratings are useful in general, but I've just become less reliant on them.

Goodhart's Law says when a measure becomes a target, then it ceases to be a good measure. In other words, when we rely on a metric so consistently over a long period of time, then it inexorably ceases to function as that metric because people will start to game it. I feel Goodhart's Law holds up fairly well across many different situations in general, where it is truer in some cases more than in others, especially as the measure becomes more widely used. It is closely related to the Cobra Effect, which dates back to an old British policy enforced to control the population of cobras in India. At time time, there were too many cobras in India, so the British government placed a bounty on dead cobras. However, this caused the locals to breed cobras so they could make even more money off of the bounty

A more recent example of Goodhart's Law is the increase in fake social media accounts, either rating restaurants well or following other users to boost their follower count. These negative externalities can even arise in A/B testing. For example, if teams are told to improve one metric (or even a few metrics) for growth, then teams can be incentivized to game that metric. A simple example of this is if a company wants to increase their views per visit, so the team paginates content into smaller pages. For more examples of how A/B testing can be gamed, refer to this blog post. As a result, metrics need to be carefully considered and monitored to make sure they're not being gamed, especially in A/B tests.

Another example of Goodhart's Law could be p-hacking. For many researchers, they need to submit scientific papers to stay in academia. Since very few esteemed journals publish negative results, then scientists can often feel pressured into producing positive results. Even without any malicious intent, they begin looking for postive results worth publishing due to this misalignment of incentives. Instead of following the scientific method by forming a hypothesis then testing it naturally, they could begin with testing many different hypotheses to see which ones might produce postive results. There becomes a chance you will find something positive (i.e. a test that satisifies the p < $0.05$ ) that is semi-similar to the original hypothesis, but isn't quite the same.

The Dead Sea Effect

Mon, 22 Aug 2022 00:00:00 GMT

Recently, I read a post from Bruce Webster's blog that highlights a pattern happening at large organizations, where the quality of retained employees exponentially worsens overtime when talented employees leave the company. In many cases, the most talented employees will be the most likely ones to leave the company if there is a workplace problem, since they are the ones most likely to secure opportunities at other companies quickly. As a result, the least talented employees are the ones that remain if there is a workplace problem. As talented employees continue to leave the organization, backfilling their roles with other talented employees becomes even more difficult. since they either notice the lack of quality right away and look elsewhere, or they will join and leave shortly afterwards. As a result, the only employees who are retained over the long-term are those that are less talented, and the degradation of talent at the large organization becomes self-reinforcing. Webster names this pattern The Dead Sea effect, since he noticed talented engineers evaporate similar to water in the Dead Sea, where less talented employees are the remaining salt residue. For additional context, water collected in the Dead Sea evaporates more quickly than water in the open ocean, making it one of the saltiest bodies of water in the world.

Comparing Common Implicit Recommenders

Tue, 15 Mar 2022 00:00:00 GMT

Since ALS was used by researchers at Yahoo in $2008$ , more recommenders have been developed to handle implicit data interactions. Many of these recommenders are based on matrix factorization and still suffer from the cold-start problem, which refers to the inability to make predictions about new customers or new items, without being forced to retrain the entire model. I've seen many different varients of the cold-start problem in different blogs and papers, but this seems to be the most prevalent definition.

Content-based models, like decision trees or clustering models, often don't experience the cold-start problem, since only the features of a new user are necessary for these models to make predictions (rather than the user ID itself). On the other hand, collaborative filtering models often are incapable of modeling a new or recently added user, since the whole model must be retrained for the collaborative filtering model to look up that user by his or her ID. For this reason, these collaborative filtering models aren't really considered to be model-based and instead are considered to be memory-based. For a more detailed evaluation of recommenders' behaviors, refer to this paper, which was published and presented at the RecSys 2014 Conference.

For the remainder of this post, I plan on illustrating the major similarities and differences between some of the most common implicit recommendation algorithms found in practice.

Comparing the ALS and BPR Losses

In practice, there are two major differences I've found between using ALS and BPR. Obviously, these differences exist because of their distinct loss functions. The first major difference relates to the ranking of items for a user based on its ratings (or relevance scores), and the second major difference relates to transfer learning happening across similar users and items built into the ratings. Note, most of the following points come from my practical experience with these recommenders, so these points may or may not perfectly align with today's literature or any future literature.

Again, the first difference between the two models relates to the degree to which a user's ranking between rated items is maintained for predicted ratings. The second difference relates to the amount of accurate transfer learning that is happening between similar items and users when ratings are predicted. When referring to collaborative filtering models, transfer learning typically refers to the degree to which a user's predicted rating for an item is based on information from other similar users and similar items. In general, models using the BPR loss function have a greater degree of transfer learning built into the predicted ratings, compared to models using the ALS loss function.

Now, let me expand on the first difference relating to ranking. ALS creates latent factors that will map back to the original rating matrix as accurately as possible, whereas BPR creates latent factors that will make sure the ranking of items for each user is maintained when mapped back to the original rating matrix. For example, let's say there is a user who gives item A a $5.0$ star rating and item B a $4.8$ star rating. Notice, this user's first and second top-rated items are close to each other in this example. ALS is more likely to create embeddings that could jumble up the ranking of these ratings, meaning the predicted rating could be $4.7$ stars for item A and $4.9$ stars for item B after the ALS embeddings are multipled together. However, BPR is more likely to create embeddings that would maintain the correct ranking of these ratings. So, the predicted rating could be $4.3$ stars for item A and $3.9$ stars for item B after the BPR embeddings are multipled together. Notice, the ranking is maintained using the BPR embeddings, but the user's ratings are more accurate using the ALS embeddings.

As with other recommenders, the BPR model can sometimes fall into a feedback loop of just recommending the most popular item. For example, BPR is more prone to recommend the most popular item if most users highly rate the most popular item and if the total number of available items is small. This point is illustrated in the example below recommending food at a completely fictional Chinese restaurant. In other words, I've noticed there is a lesser degree of newness or diversity with predicted ratings from a BPR model compared to an ALS model, since the actual item ranking is maintained much more often with a BPR model. This is a direct effect of the BPR loss function optimizing for an accurate ranking of ratings.

Now, let's expand on the second difference relating to transfer learning, which is actually a direct effect of the first difference (i.e. opimizing for more accurate ranking). According to the BPR paper, the model assumes that a user prefers their observed items (i.e. rated, purchased, viewed, etc.) over any unobserved items (i.e. items without a rating, purchase, view, etc.). For a user's unobserved items, the BPR model predicts ratings that more accurately reflect the items of similar users.

As an example, suppose most users who watch horror movies also watch documentaries, and suppose we notice a particular user has only ever watched horror movies. In this situation, there is a better chance of our BPR model predicting ratings, such that horror movies are ranked the highest and documentaries are ranked the next highest on average. On the other hand, our ALS model doesn't guarantee that documentaries are ranked the next highest. This point is especially true when the ranking of items are somewhat close together. For example, suppose users who watch horror movies nearly watch action movies as often as documentaries, but still not quite as often as documentaries. Then, ALS has a better chance of jumbling up these rankings by predicting ratings that rank action movies slightly higher than documentaries for users who watch horror movies. The reason for this happening is because BPR optimizes for accurate rankings of ratings, whereas ALS optimizes for accurate ratings.

Comparing the BPR and WARP Losses

The BPR and WARP loss functions originate from information retrieval theory and have been used in Learning to Rank (LTR) models. BPR focuses on running pairwise comparisons between samples of positive and negative items. More specifically, a BPR model involves selecting a user $u$ and an item $i$ that user $u$ deems to be positive, which could mean the user viewed the item, purchased the item, rated highly, etc. Then, BPR models will randomly select an item $j$ that the same user $u$ deems to be negative. Once the BPR model has randomly selected a positive item $i$ and negative item $j$ , it computes a predicted rating for both items by calculating the dot product of its factorized item vectors on the user vector. Meaning, the BPR model will calculate the dot product between the factorized vectors user $u$ and item $i$ , and it will calculate the dot product between user $u$ and item $j$ .

Lastly, the BPR model calculates the difference between those two predicted ratings and passes this difference through a sigmoid function. This final output is used as a weighting coefficient to update all of the model parameters using stochastic gradient descent (or SGD). Essentially, the output of the sigmoid function represents the rank of each item relative to a user, and it tells us how close or far away one item's rating is to another item's rating. In summary, we only focus on the relative ranking between items for a user, and we completely disregarded how well we predict the rating for each user-item pair.

In a similar fashion, WARP focuses on the ranking of items for each user by using a triplet loss $(user, positive item, negative item)$ , like BPR. However, the WARP loss function only updates parameters when the model predicts a negative item has a higher score than the positive item, whereas the BPR loss function updates parameters for every iteration. When iterating, the WARP loss function continues to draw negative samples until it finds a rank violation or hits some assigned threshold given as a hyperparameter. In the initial iterations, WARP makes a larger gradient update for any rank violations, indicating predictions aren't an accurate reflection of the actual ranking of items for a user. In later iterations, WARP makes smaller updates for any rank violations, since this indicates the model is producing predictions that are a more accurate reflection of the actual ranking of items. In other words, the model's predictions provide an optimal accuracy, so updates should be small.

Compared to the BPR loss function, the WARP loss function generally produces more accurate predictions (in terms of ranking), but takes more time to train since it continues to sample items until a rank violation appears. Consequently, as more epochs (or iterations) are trained, the WARP loss functions becomes much slower compared to the BPR loss function, since a violation becomes more difficult to find. Assigning a cutoff value for searching is important for training using the WARP loss function.

Most of the information comparing the BPR and WARP loss functions came from this research paper. It references the LightFM paper occasionally, which is a hybrid recommender that can use both the BPR and WARP loss functions. I recommend reading the LightFM paper for a better understanding of contextual models, the BPR loss function, and the ALS loss function. I recommend reading the research paper for a better understanding of the BPR and WARP loss functions.

Improving Health Care with 1% Steps

Tue, 01 Mar 2022 00:00:00 GMT

According to the 1% Steps Project, the three most effective ways for lowering health care costs include reducing surprise billing, capping provider prices, and providing real-time adjudication for health insurance claims. Most of the information from this post is derived from the 1% Steps Project. The motivation of this project is best summarized by the authors on the project's about page:

The goal of the 1% Steps for Health Care Reform Project is to shift the way we think about health care spending in the US and offer a roadmap to policy makers of tangible steps we as a country can take to lower the cost of health care in the US. We want to leverage leading scholars’ work to identify discrete problems in the US health system and offer evidence-based steps for reform. We will continually update the project with new proposals that are based on the latest academic research.

The Problem with Out-of-Network Billing

A surprise medical bill is an unexpected bill from a health care provider or facility. This can happen when a person with health insurance unknowingly receives medical care from a provider or facility outside their health plan’s network. Surprise billing happens in both emergency and non-emergency care. In an emergency, an individual usually goes (or is taken) to the nearest emergency department. Even if they go to an in-network hospital for emergency care, they might receive care from out-of-network providers at that facility. For non-emergency care, an individual might choose an in-network facility or an in-network provider, but they might not know that a provider involved in their care is an out-of-network provider. For example, an in-network surgeon removing wisdom teeth might use an out-of-network anesthesiologist to sedate a patient beforehand (without informing the patient). In both emergency and non-emergency circumstances, the person might not be able to choose the provider or ensure that all of their care is from a participating provider.

There are $4$ medical specialties where patients have little or no choice over the physician who treats them: pathology, emergency medicine, anesthesiology, and radiology (or PEAR physicians). As a result, PEAR physicians can refuse to join insurers’ networks, but cannot be avoided by patients. When PEAR physicians can bill out of network from inside in-network hospitals, patients can be exposed to large, unexpected, and unavoidable medical bills. In addition, the ability to engage in this profitable strategy gives these PEAR physicians the bargaining power to negotiate higher in-network payments compared to other physicians. In the end, these higher in-network payments are passed on to consumers in the form of higher insurance premiums.

In the policy brief by Cooper and Scott Martin, policy makers are suggested to improve on these problems by banning physicians from balance billing patients. To be clear, balance billing occurs when providers bill a patient for the difference between the amount they charge and the amount that the patient's insurance pays. Policy makers also are suggested to determine either the amount or the process through which out-of-network providers get paid.

In other words, there are two possible approaches. First, Cooper suggests a baseball-style arbitration approach: if an agreement between a doctor and an insurer isn't made, each would submit a bid to an arbitrator who would select between the two options for payment. Second, policy makers could require that hospitals sell a package of care with hospital and physician services at a set price, which would eliminate the possibility of a patient going to an in-network facility but being treated by an out-of-network provider.

Cooper argues regulating the nature of the contract between providers and insurers is an important potential approach in addressing out-of-network billing. He believes that price regulation is necessary since these markets are natural monopolies. Most of the information above was taken from the policy brief by Cooper and Scott Martin. For more details about out-of-network billing, refer to their other policy brief.

The Problem with Current Provider Prices

Markets for healthcare providers have become increasingly consolidated (or monopolistic). This has been happening over the last $30$ years. Since the market has become more consolidated, providers can raise prices without many cheaper alternatives. As a result, many providers will raise premiums.

Although competition drives prices to efficient levels in a well-functioning market with a healthy amount of competition, the healthcare provider market deviates from a competitive market. In particular, it deviates from a competitive in the $3$ following ways:

Reliance on health insurance shields patients from the costs of consumed services
Patients are less able to differentiate between high and low quality providers
Most hospital and specialist physician markets are highly concentrated

To expand on the first difficulty, the coverage of health insurance limits an individual's exposure to their costs of services required for a treatment. Additionally, exact costs aren't always available to the patient during their visit, since certain claims take time to be processed by the patient's insurance provider. In these scenarios, the patient must rely on pricing estimates provided by the hospital, which is based on information given by the insurance companies. These estimated charges can exclude any physician fees billed by any out-of-network surgeon, anesthesiologist, emergency specialist, pathologist, radiologist, or other physicians who may be involved in your visit. Your actual charges could be more or less than these estimates depending on specific tests ordered by your doctor, previous test results, medical history, and other factors. In summary, the current state of health insurance can lead to incorrect estimates or confusion in understanding the actual cost of healthcare services until after a patient's visit.

Second, the quality of providers is difficult to measure. As a result, consumers may have some difficulty in differentiating between high quality and low quality providers, which could lead to patients inadvertently paying for lower quality care at a higher price. Lastly, the majority of hospital and specialist physician markets are highly concentrated, which gives them substantial market power. As a result, providers can demand for higher prices due to market power alone, as opposed to a higher quality of services comparatively.

Cooper argues that pro-competition policies and regulatory intervention will improve the current state of provider prices by reducing overall consolidation. Here, pro-competition policies include vigorous antitrust enforcement and an introduction of insurance plans, which likely will incentivize patients to seek out efficient providers. Also, regulatory intervention could include enforcement of price caps, which could mean higher quality providers can charge higher prices, whereas lower quality provides can only charge lower prices. For more details about Cooper's recommendations and the estimated savings on implementing price caps, refer to Cooper's policy brief.

The Problem with Current Adjudication for Health Insurance Claims

In healthcare, adjudication is the process of reviewing and paying for a claim that has been submitted by a healthcare provider after a patient's visit and carried out by a patient's insurance company. When a patient visits his or her medical provider, the patient first will need to present his or her insurance card to a staff member at the front desk. Next, the staff member will record information from the insurance card. After the patient receives services from his or her medical provider, the medical provider will submit a claim to the patient's insurance company using the recorded insurance information. Then, the claim is processed by the insurance company and the adjudication begins.

In general, the adjudication process consists of five steps: an initial processing review, an automatic review, a manual review, a payment determination period, and a payment period. The initial processing review includes simple checks for errors or omissions, such as any incorrect spelling in the claim. Automatic review includes more detailed checks applying to the payment policies, such as checking for eligibility on the date of service or if the service should be considered medically necessary. A manual review includes checks by a medical claims examiner, which include additional checks deeming the service as necessary or unnecessary. The payment determination period involves the examiner determining if the claim should be paid, denied, or reduced, and it also involves any succeeding steps depending on the status assigned by the examiner. Lastly, the any payment will be submitted to the provider. Most adjudication (i.e. manual review) can take days several days or weeks for processing.

Administration and adjudication for claims make up about $3-6\%$ of revenues for providers and payers. These costs are driven mostly by complexity created by building on to the current adjudication process over the years. These costs are also created due to the continued reliance on manual input and review.

Each insurance payer maintains slight differences in their adjudication process. By standardizing and automating the adjudication process, the claims can be processed in real-time, the relevant administrative costs can be saved almost completely, and time can be for the insurance payer, medical provider, and patient. Additionally, price transparency can be achieved between the medical provider and insurance payer as a side effect of standardizing the adjudication process. Despite these advantages, adoption of these real-time adjudication systems remains low, mostly due to coordination failures between providers and vendors offering to implement these services.

In their policy brief, Orszag, Lazard, and Rekhi propose to solve this coordination failure through a series of interventions for insurance payers, medical providers, and relevant vendors. These interventions include the following:

Standardization of claims forms and adjudication processes across all providers and payers
New standards for reducing coding complexity
Incentives for medical providers to adopt real-time adjudication systems by providers and payers

Since the adjudication process isn't standardized across each insurance payer yet, vendors attempting to implement real-time claims processing systems experience coordination failure. For incentivizing adoption of real-time adjudication, relevant mandates could draw on authority from a range of federal programs and statutes.

For more details about the current state of adjudication and any potential improvements, refer to Orszag, Lazard, and Rekhi's policy brief about real-time adjudication for health insurance claims.

References

Outlining the A/B Testing Procedure

Mon, 28 Feb 2022 00:00:00 GMT

Experimentation using A/B testing is a crucial component in measuring customers' changes in behavior when making any changes to a business, including site changes, product changes, etc. Most of the information in this post was outlined in Dan Lee's video, which does a terrific job of defining the key steps in A/B tests. Visit his channel for more detailed A/B testing walkthroughs.

In general, Dan outlines $7$ general steps when designing and running tests, which include the following:

Defining a Problem Statement
Designing a Hypothesis Test
Designing an Experiment
Running the Experiment
Validating the Experiment and its Results
Interpreting the Results
Launching a Decision

Some software, such as Optimizely, will do some of these steps for you behind-the-scenes. For additional details about A/B testing and its benefits, refer to this article about common A/B tests outlined by a popular A/B testing software called Optimizely.

Defining a Problem Statement

Before designing any experiments or hypotheses, everyone that is a part of the test must align on a few key success metrics and an overarching goal of the experiment. Creating a user journey can be helpful in some cases as well. Metrics are created in an attempt to measure the customer or user's behavior, and great success metrics will either perfectly capture the behavior or at least be an accurate proxy for capturing the behavior. Typically, effective success metrics are crafted around the following principles:

Measurable: Will the metric track the customer's behavior using data collected throughout the experiment?
Attributable: Will the metric accurately capture any potential change in the customer's behavior when the customer is introduced to the treatment?
Sensitive: Will the metric be sensitive enough so it always will detect a change in behavior when there is an actual change?
Timely: Will the metric measure the customer's behavior in a short time window?

Each of these principles should be considered in any test when building effective success metrics. Examples of useful success metrics could include average daily revenue, average click-to-open ratios, or average click-through ratios (or CTR).

Sometimes, a metric accurately measures what a team wants to assess in an A/B test, but the metrics is determined to be insensitive. In these situations, we can use alternative metrics, try proxy metrics, and apply transformations to a metric to increase its sensitivity for a test. Some transformations include the following:

Applying a log transformation $x \to \log(1 + x)$
Capping the value at a fixed maximum $x \to clip(upper=100)$
Changing the metrics aggregation level $\frac{x}{n} \to x$

As stated already, using proxy metrics can increase sensitivity if the goal is to reach a short-term goal versus a long-term goal. Also, converting metrics into other format can increase sensitivity, such as proportions, conditional averages, or percentiles. For a more detailed walkthrough of ensuring metrics are sensitive, refer to this write-up by Microsoft about the importance of performing a sensitivity analysis, or refer to this article by Microsoft for variance reduction methods.

Designing a Hypothesis Test

After there is alignment on the problem statement and a set of business KPIs, then the testing team can design a hypothesis test using these components. Arguably, the most important piece of the hypothesis test is including the null hypothesis the alternative hypothesis. As an example, our null hypothesis might be assuming there isn't a statistically significant difference in average daily revenue between our current product ranking algorithm and our new product ranking algorithm. Additionally, the test should include the following parameters as well:

Significance level
Statistical power
Minimum detectable effect (or MDE)

Using our example, the significance level refers to the probability of observing a statistically significant difference in average daily revenue between our two ranking algorithms when there isn't actually a difference. By default, the significance level is assigned to $\alpha = 0.05$ . On the other hand, the statistical power refers to the probability of observing a statistically significant difference in average daily revenue between our two ranking algorithms when there is actually a difference. The statistical power can be evaluated by the minimum detectable effect (or MDE) in a power analysis. By default, the $MDE$ is set at $1\%$ lift, meaning a $1\%$ lift in our metric between the control and treatment group is practically significant. For detailed definitions of each of these components typically found in a power analysis, refer to the R documentation.

Designing an Experiment

After outlining the hypothesis test, our experiment should define any necessary and relevant parameters. In particular, the experiment should define basic units of measure, including who the test will randomize (e.g. buyers, email users, mobile app users, etc.). The test should also include the target population, the sample size, and the duration of the experiment. The following are some examples of relevant parameters in a test:

Test Split: 50/50
Randomization Unit: User
Target Population: US men's site visitors
Duration of the Experiment: 1 to 2 weeks
Sample Size: Number of customers based on power analysis

n \approx \frac{16 \sigma^{2}}{\delta^{2}}

Next Steps for each Hypothesis: Replace ranking system for any improvement, but do not rollout change otherwise
Success Metric stratified by Cohorts of Customers: Spend by order history, spend by gender, spend by age, spend by newness of customer, etc.

Running the Experiment

Next, the experiment must be run. Before jumping right into this step, it's important to note that the appropriate instruments and data pipelines must be set up. Additionally, the correct data, specifically any data used for success metrics, must be collected in preparation for the experiment.

Once the pipelines are implemented, the test should run for the entirety of the duration. Specifically, the test shouldn't be stopped at any point, even if a small p-value is observed early on. Obviously, the sample size will be small early on, which will cause the p-value to fluctuate in the beginning. For this reason, the test shouldn't be stopped early.

Validating the Experiment

Once the experiment is finished running, there should be initial checks made to ensure the experiment ran successfully without any bugs. Specifically, there should be guardrail metrics set up and analyzed afterwards, such as system latency times, to make sure the data was collected correctly.

Other errors and biases must be validated afterwardsm including external factors. These external factors could include less obvious biases, such as running the test during a recession, during an abnormally high spending period, during the holiday season, etc. Stratifying for external variables can help alleviate these biases, along with running ad-hoc analyses on other years or time periods for comparison.

Another important check is a validation for selection bias, which could include validating similar distributions for cohorts of users or other variables between the control and test groups. For example, we should account for a novelty effect amongst customers by segmenting customers into new and old cohorts, then stratifying them to avoid any potential selection bias. Lastly, users should be validated to make sure there is a $50/50$ split. Sometimes, a randomized experiment could lead to a $49/50$ split, which should be checked. For more examples of validations, refer to Dan Lee's video. The above validations are listed in the table below:

Bias	Check
Instrumentation Effect	Guardrail metrics
External Factors	Holidays, distruptions, etc.
Selection Bias	A/A Test
Sample Ratio Mismatch	Chi-Square Goodness of Fit Test
Novelty Effect	Segment by new and old customers

Interpreting the Results

Once the experiment and its results have been validated, the results then must be interpreted for the team who is interested in them. Results should be analyzed to ensure they're practical and statistically significant. This can be done by evaluating the results from the hypothesis tests and power analysis after running the experiment, while measuring the lifts in our designated success metrics between the control and treatment groups also. For example, p-values, relative and absolute lift differences, and confidence intervals should be reported to the team.

Launching a Decision

After communicating the results to the team, next steps should be in place to make a final decision about the tested feature. Any trade-offs should be discussed in detail before making a final decision. For exmaple, evaluating the trade-offs of observing lifts in secondary and tertiary success metrics should be taken into consideration if lifts in primary success metrics aren't observed. Additionally, evaluating the financial cost of launching the service should be considered if the lift is marginal. There are plenty of other important trade-offs to consider as well, such as the severity of any false positives arising from the test.

Intuition behind BTYD Models

Sun, 26 Sep 2021 00:00:00 GMT

In many of Peter Fader's presentations, he has demonstrated how LTV forecasts can be accurately estimated by modeling and comining two other estimates together. In particular, these two models refer to a purchase frequency model, which estimates the frequency of orders a customer will make, and a spend model, which estimates the dollar amount a customer will make. This post specificially deals with defining the former model, which is outlined in Peter Fader's paper from 2005.

This model is referred to as the Buy till you die (or BTYD) class of statistical models, which are designed to capture the purchasing behavior of non-contractual customers. Interestingly, the purchasing behavior of non-contractual customers can be summarized using two sub-models only.

The Beta-Geometric/Negative-Binomial-Distribution (or BG/NBD) model is the most popular example of a BTYD model, which estimates the purchase frequency of customers. Specifically, this BTYD model jointly models two different things:

How frequently customers make purchases while they're still alive
How likely customers are to churning in a given time period

In particular, the former process models the repeat purchasing of customers, while the latter process models the dropout rate of customers. The BG/NBD model will model the repeat purchasing process as an NBD (or poisson-gamma mixture) model, and it will model the dropout rate process as an BG (or beta-gamma mixture) model.

The Buy part (in the BTYD) refers to the frequency of orders made by each customer. Mathematically, the frequency of orders made by each customer is poisson-distributed across the entire population of customers. This distribution is parameterized by $\lambda$ , which represents the expected transaction rate made across the entire population of customers for a given time period.

The expected transaction rate comes from an underlying data-generating process representing the entire population of customers. The expected transaction rate for customers is gamma-distributed across the entire population of customers. Said another way, heterogeneity in $\lambda$ follows a gamma distribution. Implying, most customers have small transaction rates, which typically represent the lower valued customers. Some customers have larger transaction rates, which usually represent the higher valued customers. The shape of the gamma distribution (representing customer transaction rates) is defined using $r$ and $\alpha$ parameters. Here, $r$ is known as the shape parameter, whereas $\alpha$ is known as the scale parameter.

The Till you die part (in the BTYD) refers to the dropout rate for each customer. Mathematically, the dropout rate for each customer is geometric-distributed across the entire population of customers. This distribution is parameterized by $p$ , which represents the expected dropout rate made across the entire population of customers in a given time period.

The expected dropout rate comes from an underlying data-generating process representing the entire population of customers. The expected dropout rate for customers is geometric-distributed across the entire population of customers. Said another way, heterogeneity in $p$ follows a geometric distribution. Implying, customers having only made one order have a higher chance of churning, which typically represents the lower valued customers. On the other hand, customers having made multiple orders have a lower chance of churning, which usually represent the higher valued customers. The shape of the distribution representing customer dropout rates is represented using $s$ and $\beta$ parameters. Here, $s$ is known as the shape parameter, whereas $\beta$ us known as the scale parameter.

For more information about the BG/NBD model or other BTYD models, refer to Peter Fader's original paper. For higher-level illustrations about the BTYD models, refer to this helpful article. For more practical examples about BTYD models and simulations illustrating BG/NBD models in Python, refer to this article.

Potential Reasons for Recent Shortages

Thu, 09 Sep 2021 00:00:00 GMT

Recently, there have been shortages for many different products, especially in the last few years since COVID-19. The latest shortage of semiconductors has impacted multiple industries, including the automotive industry, technology industry, etc. These product shortages have been caused by recent labor shortages, port shut downs, and excess demand. Dr. Robert Handfield wrote a great article summarizing these relevant issues and their causes. Roughly, the main reasons for these particular product shortages include the following:

Preparing for a drop in demand from COVID, but instead realizing an unexpected spike in demand throughout the pandemic
A difficulty in quickly reallocating resources and capacties
Temporary closures of plants and ports due to COVID
Labor shortages in the supply chain

Let's introduce our first problem about recent shortages of semiconductors by starting off with a bit of history. Historically, semiconductor manufacturers seemed to overbuild fabricator plants and overestimate their capacities for capacity planning during successful years, which resulted in greater losses during less successful years. This happened more frequently in the 90s and hasn't happened as much recently. Since then, semiconductor companies have become more careful about capacity planning. As a result, most of them have avoided those greater losses in the last decade that they were used to seeing in the 90s.

In the last few years, semiconductor companies have been increasing capacity due to more demand related to cloud computing, machine learning, and gaming. During the pandemic, there was an increased usage of cloud computing from companies and universities. These pandemic-related spikes in demand for semiconductors weren't included in companies' planned capacity forecasts. Most likely, these companies expected drops in demand during the pandemic, rather than the realized increase in demand for semiconductors. Forecasting capacities during a time that is as turbulent as a pandemic is already difficult enough, but it becomes even more difficult since these same companies need to be extra careful about overbuilding plants, due to costly mistakes made in the past.

In other words, semiconductor shortages seem to be happening because their customers (like automotive companies) expected a reduction in demand due to COVID-19. As a result, the semiconductor fabrication plants reduced capacity and shut down their plants. With an unexpected increase in demand recently, semiconductor manufacturers have started adding another plant. However, this addition requires significant time and billions of dollars in investment, so any addition likely won't happen for another year or two.

Additionally, there have been many temporary port closures due to health and safety protocols for the pandemic. Specifically, major outbound ports, like Yantian, have been shutting down temporarily. These backups can cause many downstream complications and delays, usually requiring weeks to catch up. These shutdowns lead to a major shortage in shipping containers. These shutdowns can also cause delays in cargo ship arrivals, which also contribute to recent increases in container prices and costs of imports from China, as a result.

While all of this is going on, there are also shortages in labor relevant to these industries. For example, demand exceeds job applicants for local delivery drivers, warehouse workers, factory workers, and many other supply chain positions. Labor shortages, especially in the industries related to supply chain, are leading the slowdown of loading cargo at US ports. Again, these slowdowns lead to downstream delays and spikes in costs.

In addition, recent storms like the ones in Texas and hurrican Ida have caused chemical plants to shut down temporarily. For example, chemical manufacturing lead times were delayed for months after these storms. In particular, these chemicals are used to create compounds in many different industries, like auto parts, computers, plastics, etc. With these shutdowns, delays, shortages, and price spikes followed due to the downstream effects of necessary evacuations.

For additional information or more details about recent product shortages related to the pandemic, I recommend reading this article about reasons for recent product shortages. For more details or a brief summary about other supply chain issues related to COVID-19, I recommend reading this article summarizing recent supply chain issues.

Six Dimensions of National Culture

Mon, 23 Aug 2021 00:00:00 GMT

In the field of cross-cultural psychology, Dimensionalizing Cultures by Hofstede is still one of the most widely accepted and cited papers, even though his first paper about dimensionalizing cultures was published back in 1988. His research began when he accepted a position at IBM as a manager of personnel research, where he later founded and managed the Personnel Research Department.

In this role, he collected employee opinion surveys in over 70 national subsidiaries of IBM around the world, which eventually included over 100,000 questionnaires. This repository of questionnaires represented the largest cross-national database in existence, where he discovered significant differences between cultures in other countries. Hofstede was able to use this data to both support and contribute to pre-existing research in this field, which eventually led to the creation of six dimensions separating national culture:

Dimensions:

Collectivism (versus Individualism)
High Power Distance (versus Low Power Distance)
Masculinity (versus Femininity)
High Uncertainty Avoidance (versus Low Uncertainty Avoidance)
Long-Term Orientation (versus Short-Term Orientation)
High Indulgance (versus Low Indulgance)

At a high-level, collectivism is related to the integration of individuals into primary groups within a nation. Collectivism is arguably one of the most important axis within the six dimensions separating national culture. Additionally, power distance is related to how much less powerful members of a culture accept that power is distributed unequally. Also, masculinity is related to the division of emotional roles between women and men. Next, uncertainty avoidance is related to the level of stress that is created in a society when faced with an unknown future. Long-term orientation is related to whether a culture's efforts are more focused on the future or the present/past. Lastly, indulgence relates to the gratification of basic human desires related to enjoying life.

In general, wealthier nations are correlated with having a smaller power distance and being more individualistic. Additionally, recent economic growth has been correlated with long-term oriented countries. In his research, Triandis found that uncertainty accepting countries are slightly correlated with individualistic countries. More femanine countries tend to have a smaller power distance as countries become more and more femanine. As expected, more individualistic countries have a smaller power distance. Interestingly, there is almost no correlation between countries being masculine and countries being individualistic. Indulgant countries tend to be short-term oriented countries as well.

Most of the research for this post was motivated by a Freakonomics' Podcast about Hofstede's model, which is definitely worth a listen. For a more detailed analysis about these $6$ features, refer to Hofstede's paper about Dimensionalizing Cultures.

Illustrating the Axis for Collectivism

In 1995, Harry Triandis wrote a book titled Individualism and Collectivism, examining the influences of culture and social behaviors. At the time, Triandis' research focused on the aspects of different cultural values, and now he is considered a pioneer of cross-cultural psychology.

In this book, he frames individualism and collectivism as a single scale, which is illustrated in this paper. Essentially, collectivism is a spectrum representing the extent to which someone's personal identity is defined in terms of personal characteristics. Said another way, it's a spectrum representing the extent to which someone's personal identity is defined in terms of the characteristics of their group. An individualist just refers to someone who is relatively low on this spectrum; whereas a collectivist just refers to someone who is relatively high on this spectrum. In most cases, a person lies somewhere in the middle of this spectrum and isn't just considered one or the other. In other words, most people aren't strictly considered an individualist or a collectivist. Instead, they either lean more towards to being an individualist or a collectivist.

Keep in mind, we're all part of many different groups, including family, company, sports team, political party, country, etc. So, we'll most likely be situated at different points on this spectrum depending on the particular group we're referencing. For example, we may prefer to have more collectivistic values within our family, but we may prefer to have more individualistic values within our company.

Regardless of where someone is situated along this spectrum, both collectivists and individualists are motivated by the preferences, needs, and rights of the entity that they personally identify with. Since collectivists identify more with their collective group, they they are more motivated by the preferences, needs, and rights of their collective group. Since individualists don't identify with their collective group, then they are more motivated by their own preferences, needs, and rights.

On a similar note, both collectivists and individualists prioritize the goals of the entity that they personally identify with. Since collectivists identify more with their collective group, then they prioritize the goals of their collective group. Since individualists don't really identify with their collective group, then they prioritize their own goals.

At a high-level, collectivists refer to closely linked individuals who view themselves primarily as part of a group (or collective entity). Implying, an individual's decision is usually based on what is best for their group. Whereas, individualists refer to more self-interested individuals who view themselves as an independent entity, rathar than part of a whole. Meaning, an individual's decision is usually based on what is best for themselves.

So far, we've looked at a few effects of having an individualistic versus a collectivistic culture, but let's look at a few more. In an individualistic culture, people are expected to look after themselves and even their immediate family. On the other hand, a collectivistic culture has people who are expected to look after themselves, their immediate family, and their extended family with unquestioning loyalty.

Additionally, an individualistic culture tends to promote a right of privacy. An individualistic culture also expects individuals to have their own personal opinion, and they believe that speaking one's mind is healthy. Alternatively, a collectivistic culture tends to experience stress about belonging to a group or their culture. A collectivistic culture also believes harmony should always be maintained, rather than speaking one's mind.

Typically, collectivism prevails in less developed and Eastern countries, whereas individualism tends to prevail in more developed and Western countries. Specifically, the United States of America is ranked as the highest collectivism index. For more detailed examples about collectivistic and individualistic cultures, refer to Hofstede's Paper about Dimensionalizing Culture. The graphic below refers to a basic map illustrating Hosfstede's collectivism scores for each country, which can also be found in this article. An interactive animation can also be found here, which illustrates each score using an animation of a globe.

Illustrating the Axis for Power Distance

Power distance is defined as how much the less powerful members of a culture accept and expect that power is distributed unequally. For example, a family typically has a higher level of power distance, since children and parents expect the the parents to have a higher level of power and children to have a lower level of power. It suggests that a culture's level of inequality is endorsed by the followers as much as by the leaders. Power and inequality ultimately depend on the expectation of inequality by the followers. In almost every culture or society, there is at least some level of inequality of power, but some are more unequal than others.

In a culture with a smaller power distance, power only should be be used in a few, more legitimate scenarios, whereas power is thought to be a basic fact of life and its legitimacy is irrelevant in a culture with a higher power distance. To illustrate societies and cultures with a smaller power distance in greater detail, the following are a few examples of norms practiced in societies with a smaller power distance:

Parents commonly treat children as equals
Older people aren't outright respected or disrespected for their age
Education is centered around a teacher-enforced curriculum
Income distribution is meant to be even
Religions emphasize an equality amongst believers
Corruption is rare and scandals typically end political careers
Governments are pluralists and based on majority rule to make peaceful changes

A society or culture with a larger power distance has conflicting standards. Alternatively, the following are a few examples of norms practiced in societies with a larger power distance:

Parents teach children to be obedient, especially to their elders
Older people are respected by younger people
Education is centered around a collaboration amongst students
Income distribution is more uneven
Religions emphasize a hierarchy of priests
Corruption is more frequent and scandals tend to be covered up
Governments are autocratic and changed by revolution

In general, East European, Latin, Asian, and African countries have a higher power distance index, whereas Germanic and English-speaking Western countries have a lower power distance index. Specifically, the United States of America is ranked on the lower end of power distance indices. Refer to Hofstede's Book about Cultures and Organizations for more detailed examples about high and low power distance cultures.

Illustrating the Axis for Masculinity

Masculine cultures usually uphold values like assertiveness and competition. Alternatively, feminine cultures uphold values such as modesty and kindness. Interestingly, women's values differ less, regardless of whether the society is feminine or masculine. On the other hand, men's values can sway much more depending on their cultural identity.

For example, the women in feminine societies are quite modest and caring. In masculine societies, they are somewhat assertive and competitive, but not as much as the male population. Whereas, the men tend to be quite modest and caring in feminine societies, and they tend to be quite competitive and ambitious in masculine societies. Cultures that are more masculine tend to have the following traits:

Large differences of social and emotional roles between genders
Men should be assertive and ambitious
Women can be assertive and ambitious
Work takes a higher priority than family
Strength is an admirable quality
Fathers tend to deal with logistics and facts
Mothers deal with emotions and relationships
Women shouldn't fight physically, but men may fight physically
Women cry, whereas men don't cry
Fathers decide on the family size

Notice, the qualities found in a masculine culture tend to be synonymous with assertive and competitive values. These values in a masculine culture are often taboo, implying these values are deeply rooted, an unconscious part of a culture mostly, and rarely talked about. Feminine cultures don't tend to have such taboos, and they tend to have the following traits:

Small differences of social and emotional roles between genders
Men and women should be modest and caring
There is a balance between work and family
There is sympathy for the weaker part of society
Both fathers and mothers deal with facts and feelings
Both men and women may cry, but neither should fight
Mothers decide on number of children

In general, masculine cultures include Japan, German-speaking countries, and some Latin countries, such as Italy and Mexico. English-speaking western countries tend to have a moderately high masculinity index as well. Conversely, Nordic countries, the Netherlands, and some Latin and Asian countries, such as France, Spain, and Korea, have a low masculinity index. The United States of America is fairly masculine, since it's masculinity score is in the top 20 of all ranked countries. Refer to Hofstede's Book about Cultures and Organizations for more detailed examples about high and low masculine cultures.

Illustrating the Axis for Uncertainty Avoidance

Societies that avoid uncertainty feel uncomfortable in unstructured situations, where they have less control over the situation. Specifically, uncertainty avoidance indicates the extent to which a culture programs its members to feel either uncomfortable or comfortable in unstructured situations. Usually, unstructured situations just refer to any unknown situation that is different than the usual situation some person is comfortable with. Someone in a culture with high uncertainty avoidance is not as willing to step outside of his or her comfort zone, whereas someone in a culture with low uncertainty avoidance is more willing to step outside of his or her comfort zone. Specifically, a culture with high uncertainty avoidance has the following traits:

Uncertainty in life is thought of as a continuous threat
Uncertainty in life should be avoided
There is higher stress, anxiety, and emotion
They are less healthy subjectively
Different ideas are less tolerable and dangerous
There is a need for structure
There is a search for guidance
They stay in jobs, even if a job is disliked
There is an emotional need for rules
Religions are used to find ultimate truths and grand theories

Interestingly, uncertainty avoiding cultures are more emotional and motivated by inner nervous energy. On the other hand, uncertainty accepting cultures are more tolerant of different opinions and ideas. Uncertainty accepting cultures have the following traits:

They believe uncertainty is an inherent part of life
Uncertainty should be accepted and welcomed
They maintain higher levels of self-control and ease
They maintain lower levels of anxiety and stress
They are tolerant of new ideas and opinions
They are comfortable with ambiguity
They don't require known answers from guidance
Changing a jobs isn't a problem
There is a dislike of written and subliminal rules

Generally, east and central American countries tend to have a higher uncertainty avoidance index. On the other hand, Chinese and English-speaking cultures tend to have a lower uncertainty avoidance index. Specifically, the United States is on the tail end of the uncertainty avoiding countries. Meaning, the US tends to accept uncertainty compared to other countries. Refer to Hofstede's Book about Cultures and Organizations for more detailed examples about uncertainty avoiding and accepting cultures.

Illustrating the Axis for Short-Term Orientation

Long-term oriented societies emphasize long-term thinking and investments, whereas short-term oriented societies emphasize short-term thinking and enjoyment. Long-term oriented cultures maintain values, such as perseverance, thrift, ordering relationships by status, and having a sense of shame. On the other hand, short-term oriented cultures uphold, such as respect for tradition, reciprocating social obligations, and personal steadiness and stability. The following are traits in a long-term oriented society:

They believe the most important events in life will occur in the future
A good person adapts to mixed circumstances
Knowing what's right and wrong depends on the scenario
Traditions are adaptable to changing circumstances
Perseverance is an important goal
Large savings are available for investment
Students believe effort leads to success
Economic growth is fast

On the other hand, short-term oriented cultures have the following traits:

The most important events in life happen now
A good person will always be good
There are universal guidelines outlining what is good and evil
Traditions are sacrosanct
Family life is guided by imperatives
They're supposed to be proud of one's country
Service to others is an important goal
Students attribute success and failure to luck
Economic growth is slow

Long-term oriented countries include some Asian countries and central European countries, whereas short-term oriented countries include Middle Eastern and African countries. Specifically, the United States of America leans more towards being a short-term oriented country, with a long-term oriented index in the bottom 20 countries. Refer to Hofstede's Book about Cultures and Organizations for more detailed examples about long-term and short-term oriented cultures.

Illustrating the Axis for Indulgance

Roughly, indulgant societies represent societies that allow relatively free gratification of natural human desires related to enjoying life and having fun. Alternatively, restrained societies regulate the level of gratification by means of deep-rooted social norms. The following are traits in an indulgant society:

Many people declare themselves as happy
There is a higher importance of leisure
They're more likely to remember positive emotions
Indulgant societies usually have more educated populations
There are more people actively involved in sports
More obese people when there is enough food
There are lenient sexual norms in wealthier areas
Maintaining societal order is not a priority

On the other hand, restrained cultures have the following traits:

There are fewer people who are very happy
There is a perception of helplessness
Freedom of speech is not a primary concern
There is a lower importance of leisure
They are less likely to remember positive emotions
Fewer people are actively involved in sports
Not many obese people when there is enough food
There are stricter sexual norms in wealtheir areas
Higher number of police officers

In general, European and South American countries are more indulgant compared to other countries. On the other hand, central European countries are less indulgant compared to other countries. Specifically, the United State is ranked in the top 20 most indulgant countries. Refer to Hofstede's Book about Cultures and Organizations for more detailed examples about restrained and indulgant cultures.

Collider Bias and Police Use of Force

Thu, 20 May 2021 00:00:00 GMT

Over the last decade, I've grown up hearing and reading about police-related killings of unarmed black men, including Eric Garner, Michael Brown, Ronell Foster, George Floyd, and countless others. Naturally, there have been growing concerns about officers systematically discriminating against minorities, which has become one of the most important issues in modern-day policing. Unfortunately, this problem doesn't have such a straightforward causal answer for various reasons, if you're the ordinary layperson searching for clarity on the matter.

First, causal inference about race and policing is difficult to study due to a lack of unbiased data. Some of the most cited research papers about this topic use data provided by police forces, which suffers from confounding. For example, bias may be introduced if the data provided by polices forces are already engaging in racially-biased practices.

Throughout the last few decades, Congress has recognized this problem involving a lack of data about race and policing. In 1994, Congress instructed the Attorney General to publish annual statistics on police use of excessive force. However, this was not carried out effectively. Since then, two national systems have been built that collect data including homicides committed by law enforcement officers: the CDC's NVSS and the FBI's UCR.

The NVSS aggregates data from locally filed death certificates. However, the certificates don't document whether a killing is legally justified or not. Also, the certificates don't document whether a law enforcement officer is involved in a killing or not. On the other hand, the UCR system maintains data about police use of force. However, this relies on law enforcement agencies to submit a report voluntarily.

Clearly, governmental entities haven't made much progress in collecting unbiased and comprehensive data for police use of force, which has led to more recent efforts of data collection made by crowd-sourced projects and research institutions.

The Search for Answers

Fryer collected data in several databases, which he hoped would shed some light on police patterns. Two were public-use data sets: the New York City Stop-and-Frisk database and the Police-Public Contact Survey. The first data set includes data about the NYPD's police stops and questioning of pedestrians, if officers decide to stop and frisk pedestrians for weapons or contraband. The surveys include data about civilians describing interactions with the police, including the use of force. For more information about Fryer's data collection methodologies, please refer to this text.

Fryer's findings have been quite controversial and politicized after his paper was published in 2016. In 2020, a Princeton study disputed the findings, claiming data about white pedestrians might not be the same as non-white pedestrians, if police had a higher threshold for stopping white pedestrians. In the same year, economists from the University of Chicago published a response to Fryer's study, stating Fryer's paper "doesn't establish credible evidence on the presence or absence of discrimination against Americans in police shootings" due to selection bias. Eventually, Fryer published a response in the WSJ, reaffirming his findings as unbiased and responding to those criticisms he had received. He pointed to other recent studies from the Injury Prevention Journal and Center for Policing Equity, which have found similar results in comparison to Fryer's findings. We should note that they use the already criticized FBI data, however. Regardless, he acknowledges the existence of unchecked racial disparities today and in previous decades, and he believes his findings are enough to "justify sweeping reform."

A Glimpse into Fryer's Findings

A few facts are especially important to highlight about Fryer's study. First, Fryer finds that blacks and Hispanics are more than $50 \%$ more likely to have an interaction with police involving any use of force. In his full model using the stop-and-frisk data, he finds that blacks are $21 \%$ more likely than whites to be involved in an interaction with police where a weapon is drawn. Note, this difference is statistically significant in the study.

Once Fryer moves to the administrative data, he surprisingly finds that there are no racial differences in officer-involved shootings for police interactions. In fact, he finds blacks are $27 \%$ less likely to be shot at by police than non-black and non-hispanic suspects, after controlling for the following variables:

Suspect demographics
Officer demographics
Encounter characteristics
Suspect weapon
Year fixed effects

Note, this coefficient is measured with considerable error and is not statistically significant. In other words, Fryer is unable to use this data to find evidence for racial discrimination in officer-involved shootings.

Potential Issues with Collider Conditioning

As I mentioned already, Fryer's study generated skepticism amongst some researchers, including Dean Knox and Scott Cunningham. They suggest the administrative data sources potentially are endogenous because of conditioning on a collider. If this is true, then the administrative data itself may include a racial bias.

Fryer's study finds that minorities are more likely to be stopped in both the stop-and-frisk data and the Police-Public Contact Survey data. He introduces many controls for $X$ in the above DAG. Meaning, he's captured and controlled for hundreds of variables relating to the use of force $Y$ , which include the nature of a police interaction, time of day, etc. By controlling for $X$ , Fryer is able to close any backdoor path stemming from $X$ .

Most importantly, Knox and Cunningham point out the presence of collider bias by focusing on an officer's stop $M$ . Since the administrative data only includes data about police-initiated stops, each observation is conditional on a stop. Fryer acknowledges this:

Unless otherwise noted, all results are conditional on an interaction. Understanding potential selection into police data sets due to bias in who police interacts with is a difficult endeavor.

Conditioning on the stop variable $M$ opens up a mediated path $D \to M \gets U \to Y$ . If there is discrimination happening in $D \to M$ , spurious correlations could be created and any causal relationship between police and racial differences in shootings may be inaccurate. In other words, if there isn't racial discrimination in who the officier stops, then the correlations observed in the administrative data aren't spurious and instead induced by causal effect. Meaning, it would be fine to use this administrative data if this assumption is true. However, it would be a mistake to assume there isn't discrimination happening here, especially in a study about discrimination. As a result, we probably shouldn't make this assumption, which is in fact made in Fryer's study.

To help illustrate this point, maybe officers have a higher threshold for feeling the behavior of a white civilian is suspicious, compared to the their threshold for feeling the behavior of a minority is suspicious. Consequently, the pool of stopped white civilians would become oversaturated with more truly suspicious civilians, whereas the pool of stopped minority civilians would become undersaturated with less truly suspicious civilians. Therefore, Fryer's data first must include this missing information in order to make any accurate conclusions about the causal relationship between police and racial differences in shootings.

In a presentation at the National Institute of Statistical Sciences, Knox illustrates his point with a convincing visual provided below. He breaks down an example where a higher percentage of minority civilians are shot by officers compared to white civilians, relative to the total number of civilians in each racial community. However, by shifting our focus to stops only, a higher percentage of white civilians appear to be shot compared to minority civilians. Intuitively, this could look like more officers stopping minorities for something as mild as jaywalking, whereas white civilians are stopped only when there is more truly suspicious behavior, such as pickpocketing or robbery.

Specifically, $\frac{1}{8}$ of all white civilians are shot by officers, compared to $\frac{1}{4}$ of all minority civilians are shot by officers. But, by shifting our focus to stops, $\frac{1}{2}$ of stopped white civilians are shot by officers, compared to $\frac{1}{3}$ of stopped minority civilians. According to Knox, our conclusions could be reversed entirely depending on our analysis. In particular, Fryer's study naively concludes that there's anti-white bias because:

\frac{\text{w.s.f.}}{\text{w.s.o. } + \text{ w.s.f.}} > \frac{\text{m.s.f.}}{\text{m.s.o. } + \text{ m.s.f.}}

In order to make a conclusion about racial discrimination and policing for the entire race, Knox believes a more fair comparison would require expanding this analysis to also include non-stops for each race as well. Otherwise, Fryer's study (and others) can lead to dramatic underestimates of bias in force.

After correcting for this bias, Knox finds evidence that there is anti-minority bias by a large margin his own analysis. Specifically, Knox develops a method for correcting this bias, which places bounds on the severity of the selection problems. After applying this correction using this bounding approach, they find that even lower-bound estimates of the incidence of police violence against civilians is as much as $5$ times higher than a traditional approach that ignores the sample selection problem altogether.

For more information about Knox's study, refer to his paper for a more detailed analysis with his findings, or watch his presentation about his findings at a NISS conference. For additional illustrations about collider bias in Fryer's study, refer to this paper as well.

For additional high-level details and illustrations about the presence of collider bias in studies about policing and racial discrimination, refer to this FiveThirtyEight article.

Gram Matrices in Neural Style Transfer

Mon, 10 May 2021 00:00:00 GMT

In this paper, it has been shown that matching the Gram matrices of feature maps is equivalent to minimizing the Maximum Mean Discrepancy (MMD) with the second order polynomial kernel. Thus, the paper argues that the essence of neural style transfer is to generate a new image from white noise by matching the neural activations with the content image and the Gram matrices with the style image.

The original algorithm for neural style transfer used a cost function that minimized the sum of the content loss and the style loss. Here, the content loss represented the difference in content between the content image and our generated image. And, the style loss represented the difference in style between the style image and our generated image.

The style loss function uses the gram matrix. Specifically, the style loss represents the normalized, squared difference between the gram matrix of the style image and the gram matrix of the generated image. The gram matrix function cares about some aspects between two images, but it doesn't care about the specific presence or location of features within an image.

In the original paper for neural syle transfer, a new, generated image $x^{*}$ is iteratively created by optimizing a content loss and style loss, given by the following formula. Here, $L$ are the individual losses, and $\alpha$ and $\beta$ are the weights for content and style losses:

L_{gen} = \alpha L_{content} + \beta L_{style}

Mathematically, we can see that the losses of the generated image are just a weighted combination of the style and content losses. Here, $L_{content}$ is defined by the squared error between the feature maps of a specific layer $l$ for the the generated image $x^{*}$ and the content image $x^{c}$ :

L_{content} = \frac{1}{2} \sum_{i=1}^{N_{l}} \sum_{j=1}^{M_{l}} (F_{ij}^{l} - P_{ij}^{l})^{2}

Here, the feature maps of $x^{*}$ , $x^{c}$ , and $x^{s}$ in the $l^{th}$ layer of a CNN are denoted by $F^{l}$ , $P^{l}$ , and $S^{l}$ , respectively. Thus, the loss of the content image $L_{content}$ represents some combination of the feature maps of the generated image and content image. The loss of the style image $L_{style}$ is defined as the sum of several style losses $L_{style}^{l}$ from different layers:

L_{style} = \sum_{l} w_{l} L_{style}^{l}

Here, $w_{l}$ is the weight of the loss in the layer $l$ , and the loss of the style image $L_{style}^{l}$ is defined by the squared error between the feature correlations expressed by Gram matrices of the generated image $x^{*}$ and the style image $x^{s}$ , where the Gram matrix $G^{l}$ is just the inner product between the vectorized feature maps of the generated image $x^{*}$ in the $l^{th}$ layer.

L_{style}^{l} = \frac{1}{4 N_{l}^{2} M_{l}^{2}} \sum_{i=1}^{N_{l}} \sum_{j=1}^{N_{l}} (G_{ij}^{l} - A_{ij}^{l})^{2}

G_{ij}^{l} = \sum_{k=1}^{M_{l}} F_{ik}^{l} F_{jk}^{l}

A_{ij}^{l} = \sum_{k=1}^{M_{l}} S_{ik}^{l} S_{jk}^{l}

An Argument against Meat

Mon, 12 Apr 2021 00:00:00 GMT

Over this last year, I've grown to enjoy plant-based foods, while becoming convinced agricultural biotechnology will be increasingly important over the next decade as well. I think most people have known this need exists for livestock farming specifically, based on ethics alone. Recently, I've learned about the importance of advances in this area for other reasons as well. Below, I've listed some of these major reasons, mainly for my own knowledge.

Economics of Livestock Production

Maybe, you're someone who just appreciates efficient, well-functioning processes. In particular, you might be more interested in looking at this problem through an economic lens and learning about the sustainability of livestock production. So, let's look at the current state of animal agriculture purely from the perspective of efficiency.

Animals are inefficient converters of food, but how inefficient are they? A common metric used for measuring how much food is lost by converting animals to food is the feed conversion ratio (FCR). Typically, the FCR measures the amount of calories needed to produce one calorie of meat. Beef is one of the least efficient meats to produce. $25$ calories is required to create just $1$ calorie of beef. The ratio of pork is closer to $15$ -to- $1$ . Chicken is the most efficient meat for converting into food, and it still requires $9$ calories of input to produce just $1$ calorie of food.

Essentially, $800\%$ of what we're growing is thrown away. So, eating a plate of chicken can be thought of as simultaneously throwing out eight plates of pasta. For more information about the inefficiencies behind today's livestock farming practices, refer to this blog post by Yale's Center for Business and the Environment.

Health Effects from Livestock Farming

In 2015, the WHO called antimicrobial resistance an increasingly serious threat to global public health. More than 80% of all antibiotics sold in the United States are being fed to farm animals. Eventually, our increasingly frequent and routine consumption of meat will drive antibiotic resistance.

Antibiotics are administered to animals in feed to marginally improve growth rates and to prevent infections. In humans, there is growing evidence that antibiotic resistance is promoted by the widespread use of non-therapeutic antibiotics in animals. Resistant bacteria are transmitted to humans through direct contact with animals, which could include exposure to animal manure, consumption of undercooked meat, or contact with surfaces touched by uncooked meat.

Relevant to the recent pandemic, many previous pandemics have been caused by animal agriculture. For example, the H5N1 bird flu outbreak, which has a fatality rate of $60\%$ , originated in Chinese chicken farms in $1997$ . In $2009$ , the H1N1 swine flu outbreak likely originated in a pig confinement operation in North Carolina. In $2015$ , the H5N2 bird flu led American poultry farmers to kill tens of millions of birds containing the outbreak. Scientists expect the next pandemic to originate from a similar farming source.

For more basic nutritional value, there are healthier plant-based sources of protein, which include tofu, tempeh, lentils, and beans. Some studies, have linked plant-based diets with a lower risk of cardiovascular disease, along with a greater source of fiber and prebiotics, which improve the health of your gut.

Climate and Deforestation Impact from Animal Agriculture

Regarding climate change, raising livestock generates $14.5\%$ of global greenhouse gas emissions, where $92\%$ of the fresh water is used for farming purposes. For a more detailed overview of the effects of animal agriculture on climate, refer to this paper published by the Livestock, Environment, and Development (LEAD) Initiative.

By far, research suggests the expansion of pasture land for beef production is the biggest reason for deforestation in the Brazilian Amazon, which accounts for nearly half of forest loss in the Brazilian Amazon in $2013$ . Commercial crops, such as soy, contributed roughly $7\%$ of forest loss in the Brazilian Amazon in $2013$ .

Additionally, research suggests the large majority of global soy production is used to feed animals that will feed us later on. In a study focused on the allocation of global soy production from $2017-2019$ , around $77\%$ of all soy was used for animal feed, where nearly all of this animal feed is used in livestock farming. Only $7\%$ of global soy production is used in direct human foods like tofu, soy milk, and others. For more details about common drivers of soy consumptions, this paper published by researchers at Oxford University.

Ethics of Livestock Farming

Maybe, you're someone who just wants to learn about the treatment of a large majority of chickens. You might be more interested in looking at this problem through an ethical lens and learning about the health and wellness of livestock. So, let's look at the current state of animal agriculture purely from a moral perspective.

Since $1952$ , broiler production has been the prominent source of chicken meat, where 99.9% of chickens are raised as broiler chickens in recent years. In $2018$ , this increase in production amounted to more than $9$ billion broiler chickens produced in the United States. Today, $90\%$ of all broiler chickens are produced by independent farmers working under contract with integrated chicken production and processing companies, according to the National Chicken Council.

Over the past few decades, poultry breeding companies have repeatedly bred Cornish Cross strains to grow more white meat in the breast with less feed. So, a typical broiler chicken grows twice as fast and twice as large on half the feed compared to a broiler chicken from 70 years ago.

As a consequence, genetically engineering broiler chickens has caused their growth in weight to outpace their skeletal system and organ development. Because they put on weight so fast now, the birds’ legs and frame cannot support their bloated breasts, so they are prone to deformities. Many broiler chickens are barely able to move even at a few weeks old, due to their weight and deformities, causing them to sit and eat for most of their life. According to researchers at the University of Arkansas, these chickens grow at a rate equivalent to a two-month-old human baby weighing $660$ pounds.

Today, farms typically feed broiler chickens preventative antibiotics, creating a vicious cycle that allows them to perpetuate substandard conditions. For a more comprehensive article covering the consequences of breeding broiler chickens, refer to this story about the modern broiler chicken industry.

Understanding the Glicko Rating System

Sat, 07 Nov 2020 00:00:00 GMT

In 2007, Microsoft released their TrueSkill 1 paper, which essentially was a modified implentation of the Elo system. At the time, most of their games used Trueskill 1 as a player ranking system. Later on, the Trueskill 2 paper was released, which replaced their Trueskill 1 ranking system in most of their Xbox games, such as Halo.

Both Trueskill 1 and Trueskill 2 treat a player's ranking as a distribution, rather than a single summary statistic. To achieve this, both ranking systems use a principled Bayesian framework. As a result, the Trueskill algorithm considers the fact that all players play differently depending on their individual circumstances.

Comparing Elo with Glicko

In 1960, the Elo system was invented and used as an improved chess-rating system. The Elo system is a method used for calculating the relative skill levels of players in zero-sum games, such as chess. At the time, the Elo system became very popular in a broad range of other sports, such as basketball and football. In 1995, the Glicko system was invented to improve the Elo system by introducing a rate volatility measure. Afterward, it became a popularly adopted rating system in most well-known sports. For a more detailed explanation of the Glicko system, refer to its original paper.

The Trueskill rating systems borrow many ideas from Glicko, but also includes a measure of match quality between a set of players. There are a few other mathematical differences within the paper, and a few differences between the Trueskill 1 and Trueskill 2 rating systems. However, they share more similarities than differences. For the remainder of this post, I'll be focused on illustrating the intuition behind the Glicko algorithm.

\text{RD} = \min(\sqrt{\text{RD}^{2}_{\text{old}} + c^{2}}, 350)

As stated previously, the Glicko system extends the Elo system by computing not only a rating, which can be thought of as a best guess of one’s playing strength, but also a ratings deviation $\text{RD}$ . In statistical terminology, this $\text{RD}$ term represents a standard deviation, which measures the uncertainty of a rating. Thus, a high $\text{RD}$ corresponds to an unreliable rating, or that a player roughly has only competed in a small number of tournament games. Whereas, a low $\text{RD}$ indicates the player competes frequently.

Intuition behind Uncertainty of Ranking

In the Glicko system, a player's rating only changes based on their game outcomes, but players' $\text{RD}$ changes based on their game outcomes and their time not playing. Thus, there are two features in the Glicko system that don't exist in the Elo system. First, $\text{RD}$ increases as time passes without a player competing in many games. Second, if one player’s rating increases by $x$ , the opponent’s rating does not usually decrease by $x$ . In the Glicko system, the amount by which the opponent’s rating decreases is governed by both players’ $\text{RD}$ terms.

To apply the rating algorithm, we treat a collection of games within a rating period to have occurred simultaneously. A rating period could be as long as several months, or could be as short as one minute. As a result, an $\text{RD}$ is calculated for each rating period, which is based on $\text{RD}_{\text{old}}$ from the previous rating period.

Here, there is a constant $c$ governing the increase in uncertainty between rating periods, which can be precisely determined by optimizing predictive accuracy of future games. Once $\text{RD}$ is computed properly, the new rating $r'$ must be updated for each player.

r' = r + \frac{q}{\frac{1}{\text{RD}^{2}} + \frac{1}{d^{2}}} \sum^{m}_{j=1} g(\text{RD}_{j})(s_{j} | \text{E}(r,r_{j},\text{RD}_{j}))

\text{RD}' = \sqrt{\frac{1}{\frac{1}{\text{RD}^{2}} + \frac{1}{d^{2}}}}

The $\text{RD}'$ term refers to the updated confidence in our own ranking after playing a bunch of players. Notice, it's basically the same as our previous $\text{RD}$ , but adds a $d$ variable. Basically, the $d$ variable can be thought of as the change in uncertainty of $\text{RD}$ , or how much more certain or less certain we are about our ranking. A really large $d$ (i.e. $d \to \infty$ ) will give us the same $\text{RD}'$ , whereas a really small $d$ (i.e. $d \to 0$ ) will give us an $\text{RD}'$ of $0$ .

Essentially, the $d$ variable really starts to shrink when we play more players, which supports the fact that if we play more players, then we'll have a low $\text{RD}$ (and vice versa). So, that is how the uncertainty is updated.

Intuition behind Expected Ranking

The new actual ranking $r'$ is dependent on the difference between the outcome of an opponent $s$ and the expected outcome $\text{E}(s|...)$ of playing that opponent. This expected outcome will be around $0$ if a player is expected to lose, and around $1$ if a player expected to win. This difference is what really determines the update to a player's ranking.

\text{E}(r,r_{j},\text{RD}_{j}) = \frac{1}{1+10^{\frac{-g(RD_{j})(r-r_{j})}{400}}}

d^{2} = \frac{1}{q^{2} \sum_{j=1}^{m} g(\text{RD}_{j})^{2} \times \text{E}(r,r_{j},\text{RD}_{j}) \times (1 - \text{E}(r,r_{j},\text{RD}_{j}))}

If a player beats someone when he/she is expected to lose, this difference will be positive and high. Meaning, his/her new ranking will be relatively much higher. And, if a player beats someone when he/she is expected to win, this difference will be positive and low. Meaning, his/her new ranking won't change that much. However, if a player loses to someone he/she is expected to win against, this difference will be negative and large. Meaning, his/her new ranking will be relatively much lower. Lastly, if a player loses to someone he/she is expected to lose against, this difference will be negative and small. Meaning, his/her ranking won't change that much.

$s_{j}$	$\text{E} (s \vert ...)$	$s_{j} - \text{E}(s \vert ...)$
win	lose	$\text{much higher}$
win	win	$\text{not much higher}$
lose	win	$\text{much lower}$
lose	lose	$\text{not much lower}$

After playing a bunch of opponents, the updated ranking $r'$ is also dependent on the $g(\text{RD})$ term, which essentially is the inverse of each opponent's $\text{RD}$ term. The following formula defines this function:

g(\text{RD}) = \frac{1}{\sqrt{\frac{1 + 3q^{2}\text{RD}^{2}}{\pi^{2}}}}

q = \frac{\ln 10}{400}

Again, $g(\text{RD})$ is roughly the inverse of $\text{RD}$ . Normally, a high $\text{RD}$ indicates a player's ranking is fairly uncertain. In this case, $g(\text{RD})$ outputs a high $\text{RD}$ if a player's ranking is fairly certain. Meaning, if a player has played many games, then he/she generally will have a low $\text{RD}$ and high $g(\text{RD})$ . On the other hand, if a player doesn't play many games, then he/she generally will have a high $\text{RD}$ and low $g(\text{RD})$ .

The inverse of $\text{RD}$ represents the certainty of a player's ranking. Mathematically, we take the inverse because we want to multiply that difference mentioned earlier based on our opponent's uncertainty. So, if an opponent's ranking is highly certain, then his/her ranking will go up even more (i.e. a multiple of the difference). If an opponent's ranking is highly uncertain, then his/her ranking won't go up by very much.

Number of Games Player	$\text{RD}$	$g(\text{RD})$
many	high	low
few	low	high

Building a Prototyping Pipeline in Python

Sat, 25 Jul 2020 00:00:00 GMT

Most data science projects undergo various stages, which require communication with the business, determining use cases and opportunities with high ROIs, collecting and exploring raw data, feature engineering, and iterating through potential models.

Furthermore, these stages don't usually occur in perfectly successive steps. Rather, they generally follow an iterative cycle, where we may return to a stage after making changes to another stage. Evidently, this system consists of many moving parts, which calls for a rapid prototyping environment.

Motivating a Data Science Pipeline
Outlining a Prototyping Pipeline
Illustrating our Project Layout
Defining a Flask Application
Defining NGINX Configurations
Defining Gunicorn Configurations
Defining the Dockerfile

Motivating a Data Science Pipeline

The DevOps lifecycle delineates the journey of project development. At a high level, it determines a philosophy that enforces agility and collaboration between software development and IT operations. There are 8 phases included in the DevOps lifecycle, but this post won't be going into these specifics. For details about individual phases within the DevOps lifecycle, refer to these articles, which introduce DevOps solutions in Python.

To motivate the use of a prototyping pipeline, we'll focus on the plan and code phase for most of this post. As stated previously, there are many moving parts in a data science pipeline. Generally, any prototyping pipeline will include the following steps:

Understanding a business problem
Collecting or locating any raw data
Performing exploratory data analysis
Performing feature engineering
Building and evaluating models
Deploying ultimate model

Notice, the majority of these steps happen during the code phase, and these are only a small percentage of the complete set of operations that occur throughout the entire lifecycle. Automating these steps in a standardized environment provides benefits, such as continuous deployment, continuous testing, and process efficiency. In other words, building a rapid prototyping pipeline can help automate these steps, which facilitates project development.

Outlining a Prototyping Pipeline

When assembling my own prototyping pipelines, I personally prefer to build web applications using Flask, rather than Django. In particular, I use Dash for quickly building data visualizations and interfaces, where Dash uses the Flask micro-framework under the hood. As a result, it is fairly straightforward to embed a Dash app at a specific route of an existing Flask app.

Furthermore, a WSGI server can be used as an application server. It handles requests meant for our actual applications, which are passed on from the web server. Personally, I'll use Gunicorn for this setup, but there are many other WSGI servers that contain WSGI applications and serve HTTP.

Similarly, there are many HTTP web servers available, but Nginx is highly suggested when working with Gunicorn. Again, Nginx is our web server, which accepts the client requests and handles any HTTP connections. These HTTP requests are passed on to the Gunicorn WSGI servers. When combining these components together, our pipeline begins to take the following shape:

Although Nginx sits inside the same server as the Flask web application in this example, it can be served on its own server. Also, the web server can run in a docker container that is located within or outside of the container running our web application. In a productionized environment, we may think about running the Nginx web server to help balance the load.

In the Gunicorn documentation, the recommended number of workers is 2-4 workers per core. For horizontal scaling, Kubernetes is used to scale the number of deployments, each running a Gunicorn WSGI server with multiple workers. For more information about the implementation of Kubernetes, Gunicorn, and other components of our pipeline, refer to this post.

Illustrating our Project Layout

All files related to Docker and Gunicorn are located in the deploy directory in our project layout. The files related to our Flask web application are located in the src directory in our project layout. Ultimately, our project layout looks like the following:

myapp/
├── deploy/
│   ├── docker-entrypoint.sh
│   ├── Dockerfile
│   ├── nginx.conf
│   ├── supervisord.conf
│   └── conf.ini
└── src/
    ├── __init__.py
    └── app.py

The app.py contains the code for our web application. Whereas, the conf.py file contains configuration details for a Gunicorn WSGI server. The remaining files in the deploy directory mostly relate to Docker configurations.

Defining a Flask Application

In this post, we'll build a simple Flask web application running on Docker Compose. Specifically, the application outputs a request counter maintained in Redis. The following is sample code from our application:

# app.py
import time
import redis
from flask import Flask

app = Flask(__name__)
cache = redis.Redis(host='redis', port=6379)

def get_hit_count():
  retries = 5
  while True:
    try:
      return cache.incr('hits')
    except redis.exceptions.ConnectionError as exc:
      if retries == 0:
        raise exc
      retries -= 1
      time.sleep(0.5)

@app.route('/')
def hello():
  count = get_hit_count()
  return 'Ive been seen {} times.\n'.format(count)

In this exmaple, redis is the hostname of the redis container located on the same network as this application. We use the default port for Redis, which is 6379. This article illustrates a more detailed example using a redis task queue. For a deeper explanation about our simple Python web application, refer to the getting-started docs, which uses this example with Docker compose in greater detail.

Defining NGINX Configurations

In this post, our Nginx web server and Gunicorn WSGI server handles client requests and eventually runs our flask application as a result. Therefore, we need to configure our Nginx web server. These configurations are suggested in greater detail in the Gunicorn docs, but we'll simplify the file to only include barebones specifications in the nginx.conf file:

worker_processes 1;

pid /var/run/nginx.pid;
error_log /var/log/nginx/error.log warn;

events {
  worker_connections 1024;
}

http {
  include mime.types;
  default_type application/octet-stream;
  sendfile on;

  upstream app_server {
    server unix:/tmp/guni.sock fail_timeout=0;
  }

  server {
    listen 8080;
    server_name localhost;
    client_max_body_size 4G;
    keepalive_timeout 5;

    root /home/dkharazi/dev/myapp/public;

    location / {
      try_files $uri @app;
    }
    location @app {
      proxy_set_header X-Forwarded-For
        $proxy_add_x_forwarded_for;
      proxy_set_header X-Forwarded-Proto
        $scheme;
      proxy_pass
        http://localhost:8050;
    }
  }
}

Nginx configuration files are located in the /etc/nginx directory, where the primary configuration file refers to /etc/nginx/nginx.conf. In Nginx, configuration options are called directives, which are organized into groups known as contexts.

By default, the process ID of the nginx master process is written to the nginx.pid file in the /var/run directory. Here, we are specifying the process ID to be written to that directory, but can change it to some different directory. In a similar fashion, the error logs are written to the default directory.

The events directive specifies the context of the main configuration file. Here, we'll identify any directives related to the processing of connections. To keep things simple, we'll only specify the maximum number of simultaneous connections that can be opened by a worker.

The http directive specifies the context of the main configuration related to HTTP server directives. Here, we'll define some basic configurations, such as the default MIME type of a response, a flag for blocking I/O to disk, and WSGI server specifications. Additionally, the server directive is specified, which sets configurations for our virtual server and points Nginx to the location of our web application. These configurations redefine any appending fields to the request header, which are passed to the proxied server. Specifically, they use embedded variables, such as proxy_add_x_forwarded_for, which refers to the remote address of the client.

In our situation, we want to assign our Nginx to a port below 1024, which are privaleged ports. Also, we're informing Nginx to route all requests to the Gunicorn socket and Python application when specifying the location of @app.

For additional detils about the more basic configurations, refer to this guide. Also, refer to the docs for more details about the behavior of any particular syntax.

Defining Gunicorn Configurations

Again, our Nginx web server and Gunicorn WSGI server handles client requests and eventually runs our flask application as a result. Therefore, we need to configure our Gunicorn WSGI server. These configurations are suggested in greater detail in the Gunicorn docs, but we'll simplify the file to only include barebones specifications in the conf.ini file:

[app:server]
bind = localhost:8050
workers = 4
name = 'myapp'
daemon = True

The bind setting refers to the socket to which the Gunicorn WSGI server binds itself. The workers setting refers to the number of worker processes for handling requests. The recommended value for this setting is $2 \times \text{num cores}$ . This setting should be adjusted in order to find the best for our work load, since it depends on the hardware of our server. The daemon setting specifies that the app will run in the background on our server. For details about more specific settings in the configuration file, refer to the Gunicorn docs.

Defining the Dockerfile

Docker recommends running one process per container for these reasons. If there is a need to run our WSGI server and Nginx web server in the same container, we can use supervisord to create and manage processes based on data in its configuration file, which creates subprocesses. Docker outlines their best practices for running multiple services within a single container.

In this example, our Dockerfile builds a Python 3.7 image. We'll also define a compose file, which will reference two individual services: web and redis. Specifically, these will be outlined in the docker-compose.yml file.

FROM python:3.7-alpine
WORKDIR /myapp
ENV FLASK_APP app.py
ENV FLASK_RUN_HOST 0.0.0.0
RUN apk add --no-cache gcc musl-dev linux-headers
COPY requirements.txt requirements.txt
RUN pip install -r requirements.txt
COPY . .
CMD ["flask", "run"]

To translate some of these commands, the Dockerfile starts off with building an image using the Python 3.7 image. Then, it assigns the working directory to /myapp. Next, it sets the environment variables used by the flask command. Gcc is installed so Python packages, such as SQLAlchemy, can compile speedups. The requirements.txt file is copied, which installs the Python dependencies, and it is run after being copied. Then, all of the files from the current directory . are copied to the workdir . in the image. Lastly, the flask run command is run in the container.

As stated previously, we want to define two different services, so they can run in their own individual containers. Creating a docker-compose.yml file will make this possible. In particular, we can create a file called docker-compose.yml, which will contain the following contents:

version: '3'
services:
  web:
    build: .
    ports:
      - "5000:5000"
  redis:
    image: "redis:alpine"

The web service uses the Python 3.7 image built from the Dockerfile. On the other hand, the redis service uses the Redis image pulled from the Docker Hub registry. Port 5000 of the container is bound to port 5000 of our machine, which is the default port for the Flask web service. We can run these containers by running the following commands:

$ docker-compose up

After the command finishes executing, our container should be running. Now, we can enter http://localhost:5000 in our browser to see the application running. For additional information about specific configurations available to the Compose file, refer to the Docker docs.

Selection Bias and COVID-19

Thu, 11 Jun 2020 00:00:00 GMT

Survivorship bias refers to our tendency of only focusing on the observations that make it past some selection process, while overlooking those that do not. Typically, these observations aren't detected in some study due to some lack of visibility. Implying, survivorship bias is a type of selection bias.

This form of bias can lead to overly optimistic beliefs and conflation of correlation with causation. As an example, if the few founders of successful tech startups all dropped out of college, someone may believe that college is unnecessary or may believe tech startups have a better chance of succeeding when they're founded by college drop-outs. Obviously, we know this isn't true, since there are many students who drop out and fail, but their stories are rarely ever told.

Wikipedia illustrates many more examples of survivorship bias. Refer to it for more detailed cases involving survivorship bias, which include military and other historical examples.

State of COVID-19 Testing

As hinted at earlier, survivorship bias is a special case of selection bias, which refers to the selection of survivors. In the case of COVID-19, the current state of testing informs us about the selection of non-survivors. The CDC states on their website:

Most people have mild illness and can recover at home without medical care. Contact your healthcare provider if your symptoms are getting worse or if you have questions about your health.

In other words, those with mild symptoms are suggested to not visit the hospital, which is important for protecting health workers. Consequently, this significantly decreases the chance of ever including this group of people in the overall case count, since they're discouraged to visit the hospital early on. Furthermore, the CDC goes on to say:

An antibody test might tell you if you had a past infection. An antibody test might not show if you have a current infection because it can take 1–3 weeks after infection for your body to make antibodies. Having antibodies to the virus that causes COVID-19 might provide protection from getting infected with the virus again. If it does, we do not know how much protection the antibodies might provide or how long this protection might last.

Personally, I have tried getting a test from nearby testing centers in my city. I searched for local testing centers and found that most of them needed an appointment scheduled days in advance. Even after scheduling an appointment, nearly each test could only return results within 5-7 days. The limited testing capacities and lack of real-time reporting felt discouraging, since I essentially would need to quarantine for a week before learning if I have the virus or not. Or, I would need to drive almost an hour to a testing center much further away from my residence, and I would need to take time off from work. From someone who attempts to follow all of the CDC guidelines, this process was deflating and unrealistic.

Regarding antibody tests, I have never received one. Based on the information from the CDC, I personally would walk away from an antibody test feeling slightly unconfident with my results, given the uncertainty associated with it. Although most of this uncertainty is arguably unavoidable, these factors disincentivize symptomless people from getting tested. For more information about the potential shortcomings of antibody tests, refer to this article.

Converse of Survivorship Bias

Most likely, the factors listed above contribute to fewer asymptomatic people getting tested, which can be seen in studies like this one. The testing in hospitals tells us little about the spread of the virus because the results are prone to distortion.

Again, the survivorship bias refers to the selection of survivors. However, in the case of COVID-19, we're almost witnessing the converse of a survivorship bias. Instead, the observations in our study mostly include non-survivors, and many of the asymptomatic survivors are excluded from the case count. As a result, the virus is likely much more contagious than expected, causing the death rate to be smaller in reality. To be clear, I am not suggesting a smaller death rate means the virus shouldn't be taken as seriously. Rather, contact tracing and large-scale testing should be taken more seriously to capture a comprehensive understanding of the virus. Then, transmission models would become accurate and robust enough to better control COVID-19.

Successful countries have followed this response strategy, like South Korea. Rather than enforcing an official lockdown, they focused on aggressive testing, which specifically includes the following:

Early and frequent testing
Tracing using high-tech surveillance
Zero-tolerance isolation

As a result, South Korea is considered one of the hallmarks of the countries having contained COVID-19. For more information about South Korea's response, refer to this article

Creating Custom Awaitable Objects

Thu, 04 Jun 2020 00:00:00 GMT

The goal of the asyncio module is to implement asynchronous programming in Python. It achieves concurrency by using evented I/O and cooperative multitasking, whereas a module like multithreading achieves concurrency by focusing on threading and pre-emptive multitasking. The asyncio module focuses on coroutines, which makes this form of concurrent programming arguably more complicated than other modules, such as multiprocessing and multithreading.

When looking through the asyncio documentation, I never found any great examples that involved building custom awaitables and running them as tasks. Since so much of asyncio depends on building its own non-blocking functions specific to asyncio, this seemed strange to me. By running custom awaitables as task, we can achieve both increased flexibility and concurrency. This seems to be a very powerful component of asyncio, at least for some of my use cases.

Motivating the Await Expression

In Python 3.3, the yield from expression was introduced to wait for coroutines in asyncio applications. In Python 3.5, the await expression was introduced to replace the old yield from syntax in asyncio. It was introduced for multiple reasons and included various behavioral changes.

Compared to the yield from expression, the await syntax enforces a clearer role for coroutines. Specifically, yield from could accept a generator or coroutine, whereas await strictly accepts a coroutine.

>>> # Python 3.4 and older
>>> def foo():               # subroutine?
...     return None
>>> def bar():
...     yield from foobar()  # generator? coroutine?

>>> # Python 3.5
>>> async def foo():         # coroutine!
...     await foobar()       # coroutine!
...     return None

Introducing Awaitable Objects

In asyncio, coroutines are considered an awaitable object. There seem to be three types of awaitable objects:

A coroutine
An asyncio Task
An asyncio Future

A Future object acts as a placeholder for data that hasn't yet been calculated or fetched. A Task is a wrapper for a coroutine and a subclass of Future. Specifically, it wraps coroutines to schedule them for execution. A Task is a high-level awaitable object, whereas a Future is a low-level awaitable object. Normally, there isn't a need to create a Future object at the application level code. For these reasons, let's only focus on coroutines.

Generally, coroutines implement the __await__ special method, which return an iterator. There are a few other ways to define an awaitable object. However, each method involves defining or invoking an object with an __await__ method. Therefore, if we want to define our own custom awaitable object, we need to define a class with an __await__ special method. For a more in-depth analysis of awaitables and futures, refer to this post.

Defining an Awaitable Object

An asyncio application begins to get interesting once we start creating tasks. In asyncio, the create_task() function runs coroutines concurrently as asyncio Tasks. In this section, we'll create a task that schedules a custom awaitable coroutine.

The code below creates a custom awaitable object RandomSleeper. It sleeps for 5 to 10 seconds and returns a message after waking up. It also notifies us before falling asleep. This behavior is captured in the async def function, which creates a coroutine object. Notice, the RandomSleeper class must include the __await__ special method in order to be awaited.

As a reminder, an async def expression only creates a coroutine object once it has been awaited. Since we're interested in creating tasks, we need to create a nap function, which strictly awaits the custom awaitable RandomSleeper object. By doing this, the nap function returns a coroutine, which can be passed into our main() function.

The main() function represents an event loop, which awaits the coroutines returned by the nap() function. The main function runs these coroutines as tasks by passing them into the create_task() function.

>>> import asyncio
>>> import random

>>> class RandomSleeper:
...     def __await__(self):
...         return self.snooze().__await__()
...
...     async def snooze(self):
...         sleep = random.randint(5, 10)
...         msg = 'Sleeping for {} seconds!'
...         msg = msg.format(sleep)
...         print(msg)
...         msg = 'What a short {} second nap!'
...         msg = msg.format(sleep)
...         return await asyncio.sleep(sleep, msg)

>>> async def nap():
...     return await RandomSleeper()

>>> async def main():
...     while True:
...         t1 = asyncio.create_task(nap())
...         t2 = asyncio.create_task(nap())
...         print(await t1)
...         print(await t2)

>>> asyncio.run(main())
Sleeping for 6 seconds!
Sleeping for 10 seconds!
What a short 6 second nap!
What a short 10 second nap!
Sleeping for 9 seconds!
Sleeping for 5 seconds!
What a short 9 second nap!
What a short 5 second nap!

Notice, the two tasks t1 and t2 run concurrently in the event loop. Meaning, we're able to run our custom awaitable RandomSleeper concurrently by running its associated coroutines as tasks. Specifically, t1 and t2 are run simultaneously (roughly), and t1 waits for t2 to finish running before returning.

Internal Structure of Pandas DataFrames

Fri, 15 May 2020 00:00:00 GMT

A DataFrame object relies on underlying data structures to improve performance of row-oriented and column-oriented operations. One of these data structures includes the BlockManager. The BlockManager is a core architectural component that is an internal storage object in Pandas. Implying, it is not included in the Pandas documentation.

As the internals of Pandas continues to expand, microperformance suffers. In this case, microperformance refers to the performance of many small operations taking 1 microsecond. In particular, fairly simple oeprations, such as indexing, may pass through multiple internal layers before hitting its operation. As a result, the performance of certain operations aren't always consistent and reliable. For these two reasons alone, the BlockManager is quite important to understand when dealing with the performance of many operations in Pandas.

What is a BlockManager
Illustrating the BlockManager
Illustrating the Role of the BlockManager
Benefit of the BlockManager
Disadvantages of the BlockManager
Roadmap for the BlockManager

What is a BlockManager

In Pandas versions 0.1 and 0.2, the data in a DataFrame was stored in a dict. Since then, it has evolved into something much more complicated, but is still implemented in pure Python. Now, a DataFrame in memory roughly represents:

Some metadata
A collection of NumPy arrays for each column

This structure was introduced when the BlockManager was introduced, which manages these NumPy arrays. The reason for making this change to the structure of a DataFrame was to support column-oriented operations that were very slow without a BlockManager.

A BlockManager is fairly self-explanatory. It manages blocks, where an individual block refers to data stored as a NumPy ndarray object. The BlockManager is a memory management object that manages the internal columns of data inside a DataFrame. Each axis is capable of reshaping the blocks to a new set of labels. The BlockManager consolidates any blocks together with similar data types. It can also accept new blocks without copying data.

Illustrating the BlockManager

We may want to view the internals of a DataFrame to gain a better understanding of how the data is actually being stored. Accessing the _data attribute yields the BlockManager of a DataFrame. It also lists the specific blocks handled by the BlockManager.

>>> print(df)
   c1 c2  c3
0   1  a  10
1   2  b  20
2   3  c  30

>>> print(df._data)
BlockManager
Items: Index(['c1', 'c2', 'c3'], dtype='object')
Axis 1: RangeIndex(start=0, stop=3, step=1)
IntBlock: slice(0, 4, 2), 2 x 3, dtype: int64
ObjectBlock: slice(1, 2, 1), 1 x 3, dtype: object

Illustrating the Role of the BlockManager

As briefly described earlier, the BlockManager is responsible for consolidating any blocks together with similar data types. It does this by calling the consolidate() method.

The BlockManager doesn't consolidate blocks of similar data types when new blocks are added. Instead, the BlockManager does this automatically in the initial stages of many DataFrame operations. This notion may seem abstract at first, but can be easily observed by adding a new block to the BlockManager.

>>> df['c4'] = [100,200,300]
>>> print(df._data)
BlockManager
Items: Index(['c1', 'c2', 'c3', 'c4'], dtype='object')
Axis 1: RangeIndex(start=0, stop=3, step=1)
IntBlock: slice(0, 4, 2), 2 x 3, dtype: int64
ObjectBlock: slice(1, 2, 1), 1 x 3, dtype: object
IntBlock: slice(3, 4, 1), 1 x 3, dtype: int64

Notice, there are two separate IntBlocks after adding a new column of ints to the DataFrame. By calling the consolidate() method, we'll see consolidation of blocks of similar data types. Meaning, we'll see the two IntBlocks consolidated into one IntBlock.

>>> df._data.consolidate()
BlockManager
Items: Index(['c1', 'c2', 'c3', 'c4'], dtype='object')
Axis 1: RangeIndex(start=0, stop=3, step=1)
IntBlock: [0, 2, 3], 3 x 3, dtype: int64
ObjectBlock: slice(1, 2, 1), 1 x 3, dtype: object

Now, every block is consolidated based on its data type. At a high level, we can think of each DataFrame method calling the consolidate() method before running its operation. In truth, it is more complicated than this. Specifically, the consolidate() method is only called in operations that directly benefit from consolidation.

For a more detailed analysis about consolidation and when it happens, refer to this post. For a more detailed explanation of the BlockManager, refer to this article written by Wes McKinney, who introduced the BlockManager.

Benefit of the BlockManager

The BlockManager introduced a columnar structure to the DataFrame. Like any other columnar store, it provides significant performance boosts for column-oriented operations. Furthermore, it provides significant performance boosts to column-oriented operations on many different columns. For example, the BlockManager improves the speed of vector-like operations, such as summing two columns together.

Disadvantages of the BlockManager

Although the BlockManager was a necessary addition to the Pandas project, it creates a negative impact on performance in certain circumstances. There are four general areas that are negatively impacted by the BlockManager:

Code complexity
Loss of user visibility to memory use
Unavoidable consolidation
Microperformance issues

Since the BlockManager introduced blocks to the Pandas architecture, writing new code becomes more complex, since there needs to be careful construction of the block structure. Although this boosts the performance of complicated algorithms, such as joins, writing code for algorithms becomes more complicated.

Large datasets are usually read into a DataFrame object naively. Consequently, there is a memory-doubling effect that can lead to memory errors. When Pandas was written in 2011, the creators of Pandas weren't thinking about analyzing many gigabytes or terabytes of data. Now, the rule of thumb for reading in a DataFrame object is to have 5-10 times as much available RAM as the size of the data.

As stated previously, consolidation happens in methods that directly benefit from it. As a result, consolidation can lead to performance and memory overhead for fairly common operations. For example, calling read_csv() may require consolidation after completion.

Again, the BlockManager was a necessary addition. It fixed a lot of performance issues. However, there is a proposition to replace the BlockManager, which would require a significant inversion of the internal architecture to involve more native code and less interpreted Python.

Roadmap for the BlockManager

Currently, Pandas architecturally is structured around the following:

CPython implementation of internal data structures
Cython implementation of algorithms

In the future, there may be effort to create a native library, where the data structures, logical types, and memory management is assembled using a native API. By replacing the BlockManager with native code, Pandas would receive the following benefits:

Simpler code
Easier extensibility with new logical types
Possibly better performance than the current implementation
Improved user-control over the memory layout
Improved microperformance

For more information about the use cases and drawbacks of the BlockManager, refer to the design docs and roadmap docs, which was written by Wes McKinney after developing the BlockManager.

Performance Benchmarks: PyArrow

Sat, 11 Apr 2020 00:00:00 GMT

As of 2020, there has been development towards parquet-cpp, which is a native C++ implementation of Parquet. This development process was moved to the Apache Arrow repository.

At a very high level, the Arrow project was created primarily in an effort to provide zero-copy data access, which involves mapping complex tables to memory. Meaning, reading 1 terabyte of data from disk should be as fast and easy as reading 1 megabyte of data.

The Arrow project includes Python bindings with integration of NumPy, pandas, and built-in Python objects. These Python bindings are based on a C++ implementation of Arrow, and they are accessible via the PyArrow library. To learn more about the use cases and motivations of PyArrow, watch Wes McKinney's presentation.

The Origins of PyArrow
Benefits of the Arrow Protocol
Use Cases of PyArrow
Comparing PyArrow to Parquet
Why Serialization can be a Problem
Improvements to DataFrame.toPandas()
Setting Up Performance Tests
Performance of PyArrow

The Origins of PyArrow

In 2016, Wes McKinney joined the Apache Arrow project to improve Python's interperability with big data systems like Impala and Spark. Wes took the lead in development of the C++ and Python implementations of Apache Arrow. Rather than attempting to summarize the background of the PyArrow any further, I'll conclude this section with a quote from Wes, which is taken from an article that was written during his time at Cloudera:

At Cloudera, I started looking at Impala, Kudu, Spark, Parquet, and other such big data storage and analysis systems. Since Python and pandas had never been involved with any of these projects, building integrations with them was difficult. The single biggest problem was data interchange, particularly moving large tabular datasets from one process's memory space to another's. It was extremely expensive, and there was no standard solution for doing it. RPC-oriented serialization protocols like Thrift and Protocol Buffers were too slow and too general purpose.

Benefits of the Arrow Protocol

Apache Arrow provides an in-memory, columnar data structure that has several key benefits:

Random access is $O(1)$
Includes native vectorized optimization for analytical processing
Data interchange is fast and efficient between systems

Random access is $O(1)$ because its formatted as a column-oriented data structure. Native vectorized optimizations are possible because its execution engine takes advantage of SIMD operations, which are included in modern processors. Meaning, any algorithms that process the Arrow data structure will be very fast. Data interchange between systems is very fast and efficient, since Arrow avoids costly data serialization. Serialization is used in many other systems, including Spark, Avro, etc.

Use Cases of PyArrow

By avoiding costly serialization of I/O operations, Arrow is able improve the performance interprocess communication with zero-overhead memory sharing. Furthermore, the Arrow project involves a great deal of effort to strandardize its in-memory data structure. As a result, Arrow provides systems with the ability to reuse algorithms more efficiently.

At a higher level, Arrow tackles three general use cases:

Data movement
Data access
Computation libraries

As stated previously, Arrow attempts to efficiently improve the process of moving data from one system to another system. As a result, the Arrow memory format support zero-copy interprocess communication. In other words, reading Arrow's data structure into a separate system avoids creating any redundant data copies between intermediate buffers. Improving interprocess communication ensures RPC (client-server) based data-movement. For greater details about how Arrow achieved these benefits, read the Arrow docs

Due to improvements to interprocess communication, Apache Arrow is able to read and write to parquet files from various libraries, such as Pandas. Additionally, Arrow can convert a DataFrame object from one library to a DataFrame object from a different library. For example, Arrow is dedicating development effort to efficiently move an R data frame into Pandas, and vice versa. Arrow has already been able to improve the performance of reading a PySpark DataFrame as a Pandas DataFrame.

Arrow improves the performance of data access. These improvements include efficiently reading from and writing to common storage formats or files, such as Parquet files. Specifically, zero-copy data access enables complex memory mapping of tables, which implies accessing 1 TB of data is as fast as accessing 1 mB of data. Arrow also boosts the speed of interacting with database protocols and other data sources. Arrow also provides methods for performing efficient, in-memory, dataframe-like analytics on its data structure.

Comparing PyArrow to Parquet

My previous post described Apache Parquet as a column-oriented format for storage. It is used for data serialization and stores an actual file. On the other hand, Apache Arrow is a library that provides access to its in-memory data structure, which follows a column-oriented format. Refer to this StackOverflow post for a more in-depth explanation about the differences between these two Apache projects.

Apache Arrow defines a binary serialization protocol, which is used for arranging a collection of Arrow columnar arrays. These columnar arrays allow Arrow to provide efficient messaging and interprocess communication. The Arrow protocol is used for mapping a blob of Arrow data without doing any deserialization. This allows Arrow and other libraries to perform analytics on Arrow's data structure.

Why Serialization can be a Problem

Most Python users expect to deal with similar data structures when converting between a Spark DataFrame and a Pandas DataFrame. Usually, they want to make this conversion to use the flexible Pandas API on a locally-stored DataFrame object, after running some distribution computations on a distributed Spark DataFrame.

Users don't really care about how Spark and Pandas represent data frames internally. Rather, they have a similar data structure in mind, and they want to switch back-and-forth between Spark's API (for distributed computation on a data frame object) and Pandas API (for flexible functions on locally stored data). Since there are so many libraries that implement their own form of DataFrame object, moving from one context to another can become difficult. Particularly, loading and R data frame in Pandas can be very challenging and slow (and vice versa).

Converting from one data frame format to a different data frame format can also involve serialization. In Python, serialization refers to the conversion of an in-memory Python object, such as a DataFrame, to an on-disk stream of bytes. The cost of serialization varies for different contexts in different libraries. In particular, the pickle library is a standard Python library used for serializing Python objects. Read this for a more in depth analysis of serialization in Python.

Serialization is a relatively slow process in Python. Returning to an earlier example, a Spark DataFrame is converted to a Pandas DataFrame using serialization. Specifically, the DataFrame.toPandas() function in PySpark is inefficient, since it serializes each row into a list of tuples. This is inefficient for two primary reasons:

CPickle serialization is slow and potentially unnecessary
Iterating over each tuple using DataFrame.from_records() is a slow method for creating a DataFrame

Improvements to `toPandas`

As a reminder, the PySpark API is a very thin layer of code wrapped around the Java API for spark, which itself is only a wrapper around the core Scala API. Therefore, running a Python driver program with a SparkContext will invoke a JavaSparkContext by launching a JVM behind-the-scenes.

Arrow uses an efficient in-memory columnar data structure, which can be accessed using the PyArrow library in Python. To solve the issue with DataFrame.toPandas() in PySpark, PyArrow has proposed to use Arrow to ensure the data is in Arrow's memory format. By doing this, there isn't a need to serialize data using cPickle. This is because Arrow can send its data directly from the JVM to a Python process.

Additionally, PyArrow creates a pandas DataFrame from entire chunks of data, rather than individual values. This is achieved by using zero-copy methods in PyArrow. To learn more about the solution implemented by PyArrow, refer to this article and this post.

Setting Up Performance Tests

The New York City Taxi & Limousine Commission records each taxi and limousine trip in NYC. They report these trips to the public each month, and include information about pick-up and drop-off destinations and times. This data is used in many Data Engineering projects for reasons mentioned here.

This exercise will only use the dataset containing trips from January of 2019 completed in yellow taxis. More details about the commission and their datasets can be found on the site. This particular dataset is 687.1 MB and contains 7.6 million rows. Before performing any benchmarks, let's include any setup code:

>>> from timeit import timeit
>>> itr = 100

The timeit library is commonly used for testing performance of code segments in Python. Specifically, it returns the total seconds taken to run a given code segment, excluding the execution of any specified setup code. The iter variable is included to test each segment 100 times. After running the timeit() function, the total seconds is divided by the number of runs, ultimately to determine average performance per test.

Performance of PyArrow

As of June 2020, pyspark defaults to converting a PySpark DataFrame to a Pandas DataFrame by serializing to a cPickle. However, the toPandas() method uses Arrow when specifying the setting spark.sql.execution.arrow.enabled to true in the SparkSession.

Now, let's test the performance of the default implementation:

>>> stmnt = "df.toPandas()"
>>> setup = """
... from pyspark.sql import SparkSession
... spark = SparkSession.builder \
...     .master('local') \
...     .appName('Taxi') \
...     .getOrCreate()
... df = spark.read.csv('taxi.csv')
... """
>>> s = timeit(stmnt, setup, number=itr)
>>> round(s/itr, 2)
302.95

After performing the Arrow-less toPandas() method, the final test delivered a poorer-than-expected performance. Using the pyspark engine, converting the taxi trips DataFrame took an average of 302.95 seconds for serializing and converting the entire dataset. Now, let's test the performance of the PyArrow implementation:.

>>> stmnt = "df.toPandas()"
>>> setup = """
... from pyspark.sql import SparkSession
... spark = SparkSession.builder \
...     .master('local') \
...     .appName('Taxi') \
...     .config("spark.sql.execution.arrow.enabled", "true") \
...     .config("spark.driver.maxResultSize", "0") \
...     .getOrCreate()
... df = spark.read.csv('taxi.csv')
... """
>>> s = timeit(stmnt, setup, number=itr)
>>> round(s/itr, 2)
54.51

This example was run locally on my laptop using the default configurations for Spark. Thus, the performance benchmarks should not be taken precisely. Regardless, there clearly seems to be a huge performance boost when using Arrow. For a more detailed analysis of PyArrow's I/O performance, refer to this article.

Performance Benchmarks: Parquet

Thu, 26 Mar 2020 00:00:00 GMT

A Parquet file is a popular column-oriented storage format for Hadoop. For more information about column-oriented stores, refer to my previous post. A Parquet file is used for fast analytics that often reads and writes columns, rather than rows. Originally, Parquet files were designed to be used in MapReduce problems. Meaning, most of its development went towards parquet-mr, which is a Java implementation.

As of 2020, there has been development towards parquet-cpp, which is a native C++ implementation of Parquet. Eventually, this implementation of parquet will provide native read and write support for pandas DataFrames, which will improve the performance of:

Reading Parquet files into DataFrames
Writing DataFrames to Parquet files

The Benefits of Parquet
Compression and I/O Optimization in Parquet
The Format of a Parquet File
Setting Up Performance Tests
Performance of Parquet Engines

The Benefits of Parquet

The Apache Parquet project was originally initiated to create an open-standard columnar file format. In the beginning, Parquet files were only used in the Hadoop ecosystem. Today, they are used in Apache Spark and by cloud vendors to fill many data warehousing needs.

A parquet file is used for storing a columnar to disk. Meaning, it focuses on data compression, which refers to reducing the size of a file. In Parquet, data compression is performed column-by-column. This enables encoding schemes to be used for different data types. As a result, parquet files are able to reduce the time for each query by reducing the overall I/O, such as reading data for each column in a compressed format.

Compression and I/O Optimization

Since an entire column is stored on blocks, compression can be optimized by deducing the exact number of bits for each data value. For example, a column of integers could be compressed into a smaller data type by inferring the maximum integer value. So, if a column consist of integers that range from 0 and 100, then the column doesn't need to be any larger than int8.

I/O is optimized by focusing on projection pushdown and predicate pushdown. Here, a predicate refers to a filter with a where clause, and a projection refers to selected columns using a select clause. Projection pushdown involves column pruning. This happens automatically, since Parquet is formatted as a columnar file.

In parquet, predicate pushdown involves moving any filtering to an earlier phase of query execution. Then, it maintains statistics for groups of rows to improve the performance of predicate evaluation. In summary, predicate pushdown in Parquet provides significant performance improvements. For more details about predicate pushdown in Parquet, refer to this article.

The Format of a Parquet File

A Parquet file is organized into three general sections:

Header
Data Blocks
Footer

Each Parquet file has one header, one or many data blocks, and one footer. Within these components, a Parquet file stores two different types of information: metadata and data. Specifically, the metadata is stored in the header and footer, whereas the data is stored in the data blocks.

In particular, the header contains metadata in the form of a 4-byte magic number in the header, which represents its file is in Parquet format. Remaining metadata about the file is stored in the footer section. It contains metadata about:

Row groups
Columns
Version of its Parquet format
4-byte magic number

In a Parquet file, each data block is stored as a collection of row groups. These row groups are stored as a collection of column chunks. A row group corresponds to a set of rows, whereas a column chunk corresponds to an individual column in the dataset. The data in column chunks are organized into pages, which correspond to column values.

At a high-level, the graphic below illustrates sample data formatted as a Parquet file. For more details about the layout of a Parquet file, refer to the Apache Parquet documentation.

Use of Parquet in Pandas

As of June 2020, the pandas library provides wrapper functions that use a Parquet engine for reading and writing Parquet files. These two functions are pandas.read_parquet and pandas.to_parquet. As of June 2020, there are two choices of Parquet engines used for reading in Parquet files.

According to the pandas documentation, an engine parameter can be specified, which refers to the Parquet library to use. Its default behavior is to try ‘pyarrow’, falling back to ‘fastparquet’ if ‘pyarrow’ is unavailable.

Seeing as the development towards PyArrow and parquet-cpp is still progressing, we may be interested in performance benchmarks for reading from and writing to Parquet files, while using the above functions in their current state. For a more detailed analysis of performance benchmarks, refer to this article.

Setting Up Performance Tests

>>> from timeit import timeit
>>> itr = 100

Performance of Parquet Engines

Now, let's test the performance of reading in the same dataset in the following formats:

A csv file
An hdf file
A parquet file using the fastparquet engine
A parquet file using the pyarrow engine

Prior to executing the tests below, the HDF and Parquet files were converted to a csv file. Then, the pandas.DataFrame.to_hdf() and pandas.DataFrame.to_parquet() functions were used to store each file into their respective format.

>>> # Read csv
>>> stmnt = "pd.read_csv('taxi.csv')"
>>> setup = "import pandas as pd"
>>> s = timeit(stmnt, setup, number=itr)
>>> round(s/itr, 2)
12.92

>>> # Read hdf
>>> stmnt = "pd.read_hdf('taxi.h5')"
>>> setup = "import pandas as pd"
>>> s = timeit(stmnt, setup, number=itr)
>>> round(s/itr, 2)
6.57

>>> # Read parquet using fastparquet
>>> stmnt = "pd.read_parquet('taxi.parquet', engine='fastparquet')"
>>> setup = "import pandas as pd"
>>> s = timeit(stmnt, setup, number=itr)
>>> round(s/itr, 2)
6.64

>>> # Read parquet using pyarrow
>>> stmnt = "pd.read_parquet('taxi.parquet', engine='pyarrow')"
>>> setup = "import pandas as pd"
>>> s = timeit(stmnt, setup, number=itr)
>>> round(s/itr, 2)
3.62

After reading in each file in the various formats, the final test delivered the best performance. Using the PyArrow Parquet engine, the taxi trips dataset, formatted as a Parquet file, only took an average of 3.62 seconds for reading in the entire dataset.

Now, let's test the performance of writing to similar files.

>>> # Write hdf
>>> stmnt = "df.to_hdf('taxi.h5', key='df')"
>>> setup = """
... import pandas as pd
... df = pd.read_csv('taxi.csv')
... """
>>> s = timeit(stmnt, setup, number=itr)
>>> round(s/itr, 2)
6.18

>>> # Write parquet using fastparquet
>>> stmnt = "df.to_parquet('taxi.parquet', engine='fastparquet')"
>>> setup = """
... import pandas as pd
... df = pd.read_csv('taxi.csv')
... """
>>> s = timeit(stmnt, setup, number=itr)
>>> round(s/itr, 2)
9.12

>>> # Write parquet using pyarrow
>>> stmnt = "df.to_parquet('taxi.parquet', engine='pyarrow')"
>>> setup = """
... import pandas as pd
... df = pd.read_csv('taxi.csv')
... """
>>> s = timeit(stmnt, setup, number=itr)
>>> round(s/itr, 2)
4.96

This example was run locally on my laptop without testing other types of datasets using various styles of compression styles, such as uncompressed, snappy, and gzip. Thus, the performance benchmarks should not be taken precisely. Regardless, there clearly seems to be a huge performance boost when using the PyArrow engine. To learn more about the other use cases for PyArrow, refer to my next post.

Basics of Database Internals

Sat, 14 Mar 2020 00:00:00 GMT

A data store is a place used for storing data. This includes a database, repository, file system, etc. There are two ways of storing data in a database, which are the following:

Row-oriented data stores
Column-oriented data storees

These two types of data stores are found in many of today's popular in-memory data structures, such as pandas and Apache arrow. To see an example of a column-oriented data store, refer to my next post.

Comparing Row and Column Databases

Most of us are familar with a row-oriented data store, which stores data row-by-row. Thus, reading and writing data happens one row at a time. As a result, row-oriented data stores are used for transactional systems, which include systems that manage sales, users, airline reservations, etc. Typically, these systems read and write individual records from a row-oriented data store one at a time.

Column-oriented data stores serialize its data into columns. Column-oriented data stores are referred to as a columnar database. A columnar database stores data column-by-column. Meaning, reading and writing data happens one column at a time. As a result, column-oriented data stores are used for analytical systems, which include dashboards. Typically, these systems read and write individual columns from a column-oriented data store one at a time. In other words, the choice of a particular data store depends on the business use case.

A form for outputting user information is an example of a system that would query a row-oriented data store. It usually involves querying a database for a specified user or record, rather than an entire column. On the other hand, a dashboard outputting a graphic illustrating sales over time is an example of a system that would query a column-oriented data store. It usually involves querying a database for a sales column, rather than the sales of an individual.

Defining Data Blocks on Disk

Most databases store data on commodity hardware, which typically involves storing data on blocks. A block is the smallest unit of storage on an HDD, and it can't be partially read or written. Instead, disks read and write entire blocks of data at once. Most blocks can hold anywhere between 512 bytes to 65KB.

In application, a block typically contains multiple rows from a database. However, they are not necessarily stored consecutively. In particular, they are organized using indexing, such as B-tree indexing. Typically, indexing will physically organize data on a disk based on the logical order of the index key.

In other words, database records are stored on a data block in any arbitrary order. New records are added to any available space. When records are updated, the operating system decides their new block position. Most blocks are structured as linked lists, which contains the following:

A section for the data
A pointer to the location of the next block

Notice, each block can be scattered anywhere on the disk. Also, each block contains a pointer to the block containing the next records. The disk comes equipped with a starting and stopping index for the file as well. Refer to this post for a more detailed explanation about how database indexing works internally.

Comparing Disk I/O

As stated previously, a row-oriented database stores data row-by-row. Meaning, individual rows are stored on a block, rather than individual columns. Storing data in this manner is performant when users are querying individual rows, rather than columns, which can be seen in the following illustration:

A column-oriented database stores data column-by-column. Meaning, individual columns are store d on a block, rather than individual rows. Storing data in this manner is performant when users are querying individual columns, rather than rows, which can be seen in the following illustration:

Querying an individual record from a row-oriented database is performant, since each row is stored in a block in its entirety. Meaning, the disk only needs to read from or write to the record in one place. On the other hand, querying an individual record from a column-oriented database will not be as performant. This is because it takes $O(n)$ to read from or write to each block for a piece of the row.

Querying an individual column from a column-oriented database is performant, since each column is stored in a block in its entirety. Meaning, the disk only needs to read from or write to the column in one place.

In the images above, each sector represents a single block for illustrative purposes only. Making this oversimplification hopefully provides better intuition of the purpose of blocks, since the process of reading from and writing to blocks becomes simplier. For a more in-depth explanation about some of these concepts, refer to this video.

NoSQL Basics: Graph Databases

Sun, 01 Mar 2020 00:00:00 GMT

In a previous post about NoSQL databases, graph stores were described at a fairly high-level. In this post, we'll dive into more low-level details, which includes features, behavior, and use cases.

NoSQL Distilled is a terrific resource for learning about both high-level and low-level details of NoSQL databases. This post is meant to summarize my experience with these databases, along with particular segments from the book. Again, refer to the book for a deeper dive of relational and NoSQL databases.

Defining a Graph Database
Features of Graph Databases
Using Graph Databases

Defining a Graph Store

Graph databases store two types of objects:nodes and relationships. These nodes represent data entities, and replationships represent a relation between two entities. These relationships visually are represented as edges between two nodes. Each edge between entities has a direction and properties as well. In the world of graph databases, edges can be thought of as a series of joins.

Graph databases were designed to handle data with many relatonships. Roughly, relational data involving many different joins may be better suited in a graph database, since relationships are stored more efficiently. Graph databases intuitively manage relationships better than relational databases, since relationships are stored as objects themselves. As a result, lookups between tables for person ID and department ID don't need to be performed constantly, in order to find which person connects to which department. In other words, the relationships don't need to be inferred anymore.

This ability to pre-materialize relationships into the database structure allows Neo4j to provide performance of several orders of magnitude above others, especially for join-heavy queries, allowing users to leverage a minutes to milliseconds advantage.

Neo4j offers a declarative graph query language, which is built on the basic concepts and clauses of SQL. Additionally, there are many other functions to simplify the process of working with graph models. Here is a SQL query:

SELECT name FROM Person
LEFT JOIN Person_Department
  ON Person.Id = Person_Department.PersonId
LEFT JOIN Department
  ON Department.Id = Person_Department.DepartmentId
WHERE Department.name = "IT Department"

And this is the corresponding Cypher query:

MATCH (p:Person)-[:WORKS_AT]->(d:Dept)
WHERE d.name = "IT Department"
RETURN p.name

Notice, the query simplifies the heavy-use of joins. Not only does Neo4j simplify this process, it also improves the performance of join-heavy queries compared to SQL. The relationship between nodes is not calculated at query time, but it actually is persisted to disk. In relational databases, adding another relationship involves intricate changes to the schema. However, this becomes much simpler in Neo4j.

For a more detailed comparison between graph and relational databases, refer to the Neo4j docs.

Features of Graph Databases

Similar to relational databases, most graph databases are ACID-compliant. Meaning, only valid transactions are individually committed to the database, even in the event of a failure. If a transaction isn't marked as finished or successful, then it will be rolled back.

Most graph database solutions aren't able to be distributed across multiple servers, favoring consistency and availability. A few graph databases, such as Infinite Graph, support node distribution across a cluster of servers, such as Infinite Graph. Whereas, Neo4j is implemented using a master-worker architecture that is fully ACID-compliant.

As emphasized already, graph databases handle data with complex relationships quite efficiently, since relationships are indexed themselves. Therefore, an RDBMS should be preferred if we're mainly interested in filtering individual entities. Graph databases should mainly be used if we're querting relationships.

Graph databases are useful for reading relationships, rather than writing and rarely reading these relationships. Specifically, other databases should be preferred if we're looking to write data and don't expect to query our stored entities or relationships often.

As opposed to relational databases, graph databases are fairly robust for data that is constantly changing. Specifically, graph databases can add relationships and properties, since graph databases can change traversing requirements without having to change its nodes or edges.

In other words, relational databases may be a preferred choice if the columns of a table aren't expected to change. Otherwise, graph databases may be a preferred choice if the requirements are expected to morph over time.

Graph databases are useful when searching throughout graph for a particular relationship. A graph database is optimized for traversing the graph for a relationship. If we're mainly interested in querying entities without a specific relationship in mind, a relational database may be better suited for our needs.

As an example, a graph database is useful when we're wondering what people Jennifer knows. On the other hand, a graph database may not be very useful when we're wondering who Jennifer knows. A good query for a graph database is:

MATCH (n:Person {name: 'Jennifer'})-[r:KNOWS]->(p:Person)
RETURN p

Since the map would need to be traversed entirely, a poor query for a graph database is:

MATCH (n)
WHERE n.name = 'Jennifer'
RETURN n

It should go without saying, but graph databases don't perform well when used as key-value stores. A standard lookup operation is best used for a key-value database or even a relational store. Graph databases are more useful for narrowed-down relationship lookups of a key.

Graph databases are performant when reading and writing smaller nodes. Graphs are not suited for storing many properties on a single node. They are also not suited for storing large values within those properties. This is because the query can hop from entity to entity quickly. However, graph databases need extra processing to pull out details for each entity along a search path. For additional summaries about features within graph databases, refer to this article.

Using Graph Databases

The table below outlines a few particular use cases for graph databases. In particular, the graph databases use Neo4j as a representative of the following use cases. Read more details about the Neo4j use cases in the docs. For more details about individual use cases, refer to the NoSQL Distilled text.

Use-Case	Good or Bad?
Connected Data	Good
Social Network Data	Good
Delivery Routing	Good
Location-Based Services	Good
Recommendation Engine	Good
Fraud Detection System	Good
Updating all or some entities	Bad
Complicated operations	Bad
Large data	Bad

NoSQL Basics: Column-Family Databases

Thu, 27 Feb 2020 00:00:00 GMT

In a previous post about NoSQL databases, column-family stores were described at a fairly high-level. In this post, we'll dive into more low-level details, which includes features, behavior, and use cases.

Defining a Column-Family Database
Introduction to the Cassandra Architecture
Features of Column-Family Databases
Using Column-Family Databases

Defining a Column-Family Store

Compared to key-value and document databases, column-family stores impose more limitations on the structure of an aggregate. Specifically, column-family databases organize their columns into column families. Each column must be assigned to a particular column-family. Then, each column can be accessed via a column-family. In particular, accessing a column-family will return each of the columns associated with that column family.

At a high level point of view, a column-family database represents a map consisting of smaller maps, where the first map is a column family and the second map is a row. In the CQL API, column families are referred to as tables. To gain some additional intuition about the data model, let's look at how data is stored in Cassandra:

-- create keyspace (database)
CREATE KEYSPACE hotel WITH replication =
    {'class': 'SimpleStrategy',
     'replication_factor': 3};

-- create table
CREATE TABLE hotel.employees (
    id text PRIMARY KEY,
    name text,
    roles set,
    salary smallint
)

-- insert row
INSERT INTO hotel.employees
       JSON '{"id": "10F-S53",
              "name": "Susan",
              "roles": {"accountant", "auditor"}}';

CQL is a typed language used for querying column-family databases in Cassandra. Meaning, Cassandra isn't really schemaless anymore, since data types and primary keys are required. In many ways, CQL feels like SQL, but deliberately excludes certain functionalities that violates their data model. In particular, CQL doesn't support group by operations, join operations, and others.

On the flip side, CQL supports certain functionalities that aren't supported in relational databases, since they violate the relational data model and comply with the column-family data model. For example, CQL supports the use of tuples and sets as data types defined within a schema.

To learn more about the details behind data manipulation with CQL, refer to the docs.

Introduction to the Cassandra Architecture

Whereas Apache HBase was created based on Google's BigTable, Apache Cassandra relies on a number of techniques from Amazon's Dynamo. For a brief comparison between HBase and BigTable, refer to this post. Each node in the Dynamo system has three main components:

Request coordination for each partitioned dataset
Ring membership and failure detection
Local storage engine

Cassandra uses most of these features, but uses a storage engined based on LSM instead. To go one level deeper, Cassandra takes the following from Dynamo:

Dataset partitioning using consistent hashing
Multi-master replication
Tunable levels of replication and consistency
Distributed cluster management
Distributed failure detection
Incremental horizontal scaling on commodity hardware

Cassandra partitions (creates shards) data across nodes using consistent hashing. In naive data hashing, keys are allocatd to buckets by hashing the key modulo the number of buckets. Cassandra takes a different approach by first hashing each node to one or more values on a continuous hash ring. These hash values representing each node are referred to as tokens in Cassandra. Once tokens are created, Cassandra then is able to map data points to tokens on that same hash ring. Specifically, Cassandra will receive rows, hash the primary keys of each row, and map those hash values to the hash ring. Lastly, Cassandra will map those data points to nodes by rounding their mapped hash values to the nearest token in a clockwise motion on the ring.

The use of consistent hashing for partitioning makes Cassandra a scalable and available column-family store. There are other features included in the hashing algorithm to improve potential issues with consistency, such as virtual nodes, quorums, and compaction. For more details about the architecture of Cassandra and its more specific hashing features, refer to the docs.

Features of Column-Family Databases

Unlike some NoSQL databases and most relational databases, Cassandra does not support transactions. Similar to key-item and document stores, wrties are atomic at the row level in Cassandra. Meaning, any transformation of multiple columns of a single row is treated as a single write operation. For more information about atomicity in Cassandra, refer to the docs.

Returning to the CAP theorem, Cassandra focuses on high availability and scalability. The consistency can be increased with the use of quorums. Each quorum has a replication factor, which can be adjusted to tune the level of availability within our cluster.

Whereas both key-item and document databases are designed using a master-worker architecture, Cassandra uses a peer-to-peer architecture. This choice reaffirms the idea of Cassandra favoring availability and scalability, rather than consistency.

Cassandra effortlessly scales with the addition of nodes. Since Cassandra doesn't promote a master node, it really doesn't need to worry about failures or spend time promoting a new master node.

Lastly, Cassandra supports a fairly robust querying language called CQL. Since Cassandra isn't a relational databases, CQL still has its own limitations compared to SQL. Thus, column families will need to be designed effectively, so they are optimized for reading data.

Using Column-Family Databases

The table below outlines a few particular use cases for column-family databases. In particular, the column-family databases use Cassandra as a representative of the following use cases. Read more details about the Cassandra use cases in this post. For more details about individual use cases, refer to the NoSQL Distilled text. For a more straightforward comparison between MongoDB and Cassandra, read this article.

Use-Case	Good or Bad?
Event logging	Good
Blogging sites	Good
Content management systems	Good
Real time analytics	Good
Page counters	Good
Systems requiring ACID	Bad
Early prototypes	Bad

NoSQL Basics: Document Databases

Sat, 22 Feb 2020 00:00:00 GMT

In a previous post about NoSQL databases, document stores were introduced at a fairly high-level. In this post, we'll dive into more low-level details, which includes features, behavior, and use cases.

Defining a Document Database
Introduction to MongoDB
Features of Document Databases
Using Document Databases

Defining a Document Store

A document database is a strongly aggregate-oriented database. Meaning, it consists of many aggregates. In particular, document databases generally enforce fewer restrictions and offer increased flexibility, compared to both relational databases and other NoSQL databases.

Document databases generally enforce similar constraints as key-value databases. However, they can sometimes enforce a few additional limitations, such as:

Size limits
What we can place in them
Data types

Additionally, the fields in document databases can be queried, which contrasts with key-value databases. In other words, we can query specific segments of an aggregate in document databases, whereas we can only query the entire aggregate (belonging to a key) in key-value databases. Lastly, indices can be created based on the contents of an aggregate in a document database.

Obviously, a document database stores documents, which are hierarchical tree data structures. These data structures can be XML, JSON, etc., and they can consist of scalar values, collections, and maps.

At a high level point of view, document databases can be thought of as queryable key-value databases. In key-value databases, aggregates are only accessible by means of a key. As a result, values can't be queried and requires a lookup for the entire aggregate. Contrastingly, document databases allow their values to be queried.

Each document in a document database can include different attribute names. In other words, the schema for each document can be different across documents. Whereas, each row in a schema must follow the same schema for each table.

Introduction to MongoDB

MongoDB is a document database that stores data in JSON-like documents. Again, document databases have many similarities with key-value databases. In particular, key-value databases store two objects: a key as a string and a value as a string, list, etc. Whereas, document databases store a single object: a JSON-like document.

Not only does MongoDB allow developers to store their in-memory data structures in a straightforward way, but it also provides a comprehensive suite of tools useful for analytics. For one, MongoDB automatically creates charts of any MongoDB data stored inside the database. It also leverages external BI tools via connectors, such as Tableau, Qlik, and others.

MongoDB also provides features, such as data searches and visualizations, via an intuitive GUI. By doing this, developers are able to manipulate data with visual editing tools.

Features of Document Databases

In MongoDB world, tables are referred to as collections. Thus, a document is stored in a collection within a database. In an earlier post, the benefits of document stores were briefly mentioned, which include consistency and flexibility benefits. Specifically, each document doesn't care what data is stored as their values. The values can be JSON, XML, etc. The structure of the data model is flexible.

The CAP Theorem implies relational databases ensure strong consistency, which means any reads sent to a relational database will always return the most recently written value. Strong consistency is simpler to achieve with relational databases, since it typically consists of a single node.

On the other hand, document databases ensure eventual consistency. Recall, if two reads happen after a write in a distributed system, they may access data from two different nodes. Eventual consistency implies those reads will receive the same values shortly after the write. However, eventual consistency also implies those reads may not receive the same values if the reads happen too soon after the write. In this scenario, replica nodes need more time to receive the most-recent value written to the master node.

Document databases support workarounds to increase availability. One approach for improving availability involves the use of a replica sets. A replica set is a group of processes that maintain the same dataset. Replica sets replicate data to provide redundancy and high availability.

By using replica sets, consistency can also be tuned in a way that waits for writes to be replicated to a certain number of replicas. Then, every write will ensure that number of replicas are written, before any value is returned successfully. Similarly to key-value databases, increasing the consistency will slow down writes, since more nodes need to be propagated.

Document databases, such as MongoDB, is implemented using a master-worker architecture, which means all requests are sent to an individual master node. Then, the data from that master node is replicated to its worker nodes. If the master node ever experiences downtime, the worker nodes will vote for a new master node among themselves.

Whereas key-value stores can only be queried via their key, document databases support limited querying capabilities based on attribute values in a document. As an example, MongoDB provides very simple querying oprations for filtering and ordering.

Lastly, many document databases are scalable using sharding. Sharding introduces its own of issues involving problems with availability and complexity. However, these tradeoffs can be tuned using a parameter representing the number of node failures.

Using Document Databases

The table below outlines a few particular use cases for document databases. In particular, these document databases use MongoDB as a representative of the following use cases. Read more details about the MongoDB use cases in the MongoDB docs. For more details about individual use cases, refer to the NoSQL Distilled text.

Use-Case	Good or Bad?
Event logging	Good
Blogging sites	Good
Content management systems	Good
Analytics	Good
Applications needing a flexible schema	Good
Multioperation requests	Bad
Many differently structured documents	Bad

NoSQL Basics: Key-Value Databases

Sun, 16 Feb 2020 00:00:00 GMT

In a previous post about NoSQL databases, key-value stores were described at a fairly high-level. In this post, we'll dive into more low-level details, which includes features, behavior, and use cases.

Defining a Key-Value Database
Introduction to Redis
Features of Key-Value Databases
Using Key-Value Databases

Defining a Key-Value Store

A key-value databas is a strongly aggregate-oriented database. Meaning, it consists of many aggregates. In particular, key-value databases generally enforce fewer restrictions and offer increased flexibility, compared to both relational databases and other NoSQL databases. In certain situations, there enforce minor constraints, such as size limits. However, they generally offer more freedom comparatively.

In key-value databases, aggregates are only accessible by means of a key. As a result, values can't be queried and requires a lookup for the entire aggregate. Without the use of a strict schema, any data can be stored under a key in a key-value model.

Essentially, key-value databases are just hash tables. They are useful for storing data that interacts with an API. Indicating, key-value stores are useful for clients who only need to do the following:

Add a key-value combination
Get the value of a key
Store a value for a key
Delete a key

In key-value databases, values are stored as blobs, which implies these values roughly can be represented as data type. Accessing values associated with a key is both scalable and performant, since there is only one way to access the aggregate. Moreover, understanding what value is stored falls on the shoulders of the application developer.

Introduction to Redis

Redis doesn't only accept primitive data types for its values. It supports data structures such as strings, hashes, lists, sets, sorted sets, bitmaps, hyperloglogs, geospatial indexes with radius queries and streams.

Redis supports operations that include appending values to a string, pushing elements to lists, unions, etc. These features make Redis a popular choice when deciding between the many key-item databases. Redis achieves its performance benefits by working with in-memory datasets. Datasets can be persisted to disk by either occasionally saving them to disk or logging any executed commands.

For many of the reasons listed above, Redis is mostly used for storing in-memory data structures, which happens often in caching and messaging system. Its features make Redis a popular choice when deciding between the many key-item databases. Its flexibility and atomic operations separate Redis from the others. For a more detailed explanation of Redis, refer to their site.

Features of Key-Value Databases

In an earlier post, the benefits of key-value stores were briefly mentioned, which include consistency and flexibility benefits. Specifically, each key doesn't care what data is stored as their values. The values can be JSON, XML, text, etc. The structure of the data model is flexible.

On the other hand, key-value databases ensure eventual consistency. Recall, if two reads happen after a write in a distributed system, they may access data from two different nodes. Eventual consistency implies those reads will receive the same values shortly after the write. However, eventual consistency also implies those reads may not receive the same values if the reads happen too soon after the write. In this scenario, replica nodes need more time to receive the most-recent value written to the master node.

Key-values databases support workarounds to increase consistency, but some of these solutions decrease write performance. One approach for improving on consistency and write tolerance involves the use of a quorum. Quorums involve setting a replication factor, which tolerates nodes being down for write operations.

As stated previously, key-value stores can only be queried via their key. This constraint is one of the reasons for their excellent performance, but may be problematic if we don't know the key. Therefore, there is an important design consideration that needs to be taken when creating the keys.

Lastly, many key-value databases are scalable using sharding. Sharding introduces its own of issues involving problems with availability and complexity. However, these tradeoffs can be tuned using a parameter representing the number of node failures.

Using Key-Value Databases

The table below outlines a few particular use cases for key-value databases. In particular, these key-value databases use Redis and Riak as representatives of the use cases. However, these two key-value stores have their own separate use cases from each other. Read more details about the Redis use cases in the Redis docs, and Riak use cases in the Riak docs. For more details about individual use cases, refer to the NoSQL Distilled text.

Use-Case	Good or Bad?
Session caching	Good
User profiles	Good
Configurations	Good
Shopping carts	Good
Relationships	Bad
Multioperation requests	Bad
Querying by values	Bad
Operations applied to many keys	Bad

Relational and Non-Relational Databases

Sun, 09 Feb 2020 00:00:00 GMT

Ensuring a stable form of data storage is an important decision for any business. The data in an organization can last much longer than many of its applications. Unfortunately, there isn't a single database that solves the needs for every business. Each database has its own set of properties, and an organization can utilize those properties to find the database (or databases) that best solves their distinctive problem.

Evaluating Databases with CAP Theorem
Distribution and Replication
Power of Relational Databases
Disadvantages of Relational Databases
What is NoSQL
Purpose of NoSQL Databases
Properties of NoSQL Databases
Introducing Data Models in Databases
Types of Aggregate Data Models
Use Cases of Database Types

Evaluating Databases with CAP Theorem

At a high level, there are a few categories that a database can fall under. First, there is the relational database, which is one we're all most likely familiar with. Relational databases are organized into tables with columns and rows. They ensure transactions are ACID and enforce many restrictions using a schema. ACID is a somewhat contrived acronym that will be explained in greater detail later.

Additionally, there are the non-relational databases, which offer increased flexibility by embracing schemaless data. Unlike relational databass, there are many different flavors of non-relational databases, such as key-value, document, column-family, and graph databases. As stated previously, each of these databases are schemaless, and the graph database is the only one ensuring ACID transactions.

After introducing a few modern-day database options, we now should think about how to evaluate databases that are best suited for our organization's needs. To do this, we'll consider the CAP thorem, which assesses the tradeoffs between three metrics:

Consistency
Availability
Partition-tolerance.

At a high level, the CAP theorem states we can't choose two of the above metrics without sacrificing the third metric. A partition refers to an interruption in communication within a distributed system. Thus, partition indicates if any node goes down, then the cluster will still be up. Availability ensures a request to any node that is up will return a valid response. Consistency implies a request to any node that is up will return the same response. For more information about the CAP theorem, refer to this IBM article.

By observing the image below, we'll notice relational databases maintain both availablity and consistency. On the other hand, NoSQL databases can maintain either consistency and partition-tolerance, or they can maintain availability and partition-tolerance. If we're certain that our organization only ever will need one server, then we won't need to worry about partition-tolerance, and we can prioritize both availability and consistency. On the other hand, if we're certain our database will require more than a single server, then we may need to choose between consistency and availability.

The CAP theorem has become controvertible over the years. In practice, these tradeoffs have become more loose for some of these databases. Using Cassandra as an example, the inclusion of a quorum almost allows the amount of consistency to be configurable. Over time, the lines separating the databases from each other are becoming blurred. Many of these databases can be made to work in any situation, but we should choose a database that is built for our requirements. A more detailed explanation about other considerations can be found here.

Distribution and Replication

In NoSQL databases, there are two basic styles of distributing data. First, there is sharding, which has been mentioned in a previous post. Sharding distributes different data across multiple servers, so each server becomes the single source for its subset of data. In other words, sharding involves splitting up a dataset into sections, then storing each section on its own server.

Replication is the second form of distributing data. Replication copies data across multiple servers, so there are multiple copies of the same dataset stored in more than one place. In other words, replication involves duplicating a dataset on another server. A system may use neither of these techniques, one of these techniques, or both of these techniques.

To go one level deeper, there are two forms of replication. First, a master-worker form of replication promotes one node as the authoritative copy. In this architecture, the master typically handles writes, while the workers handle reads. Consequently, this choice causes master-worker architectures to be eventually consistent, since it takes time for the written values to be updated on the workers for reading.

Secondly, a peer-to-peer form of replication doesn't use a master. Instead, it allows writes to any node, so each node coordinates with each other to synchornize their copies of data. In general, master-worker replication reduces the chance of update conflicts, but peer-to-peer replication avoids loading every write onto a single node, becoming a single point of failure.

Most systems will need to choose one form of replication over the other form. Many distributed systems promote a combination of both sharding and replication. Also, most of the terminology mentioned above is defined in the NoSQL Distilled textbook, so please refer to it for more details.

Power of Relational Databases

Before introducing a few significant properties and use cases for relational databases, we should briefly review why we're intersted in writing to a database in the first place. Most architctures have two general ways of storing data: writing to memory or writing to disk. Keep in mind, data remains persistent by writing to disk.

In other words, any data stored away in memory is lost if we lose power or observe any hardware issues. For these reasons, data needs to be written to disk in order to ensure persistence. Any data written to disk is commonly accessible to us via files in a file system on our operating system, or via a database.

Since both approaches are ways to access data on disk, one of the first questions to be asked is can we just keep our data in the file system, rather than writing it to a database?

This is a valid question. However, an organization typically will require the use of a database, which relates to the increased flexibility of a database in storing large formats, compared to file systems.

In relational databases, a request sent to the database is known as a transaction. Relational databases control users accessing its data via transactions. With the use of transactions, a relational database can undo a change if that change causes an error, without creating issues for other transactions being performed simultaneously. In particular, transaction are run in an isolated environment, allowing for other transactions to execute concurrently.

Relational databases support manipulation of multiple rows in a single transaction. For relational databases, these transactions are ACID transactions:

Atomic
Consistent
Isolated
Durable

Atomicity implies an entire transaction will fail even if only one part of a transaction fails, and there won't be any changes committed to the the database afterwards. Consistency implies any successful transaction will take the database from one valid state to another valid state. Isolation implies any concurrent operations would have the same results if they were executed serially. Durability implies any transaction must remain committed even in the event of a crash or power loss.

The major takeaway from the ACID acronym is atomicity. Meaning, updates to multiple rows from different tables would be updated as a single operation, which implies an operation either succeeds or fails in its entirety. By enforcing this rule, concurrent operations are isolated from each other, so they'll never create a partial update.

Across many different types of relational databases, they maintain a fairly standard model. In general, transactions operate fairly uniformly, regardless of which relational database is selected. In addition, the SQL dialects for most relational databases are fairly interchangeable. For example, Microsoft SQL Server and MySQL both include fairly similar querying languages.

Disadvantages of Relational Databases

Technically, a relational database consists of tuples. A tuple is synonymous with rows, but are organized as name-value pairs. A relation is referred to as a set of tuples. All SQL operations in relational databases consume and return relations. These relations represent output of mathematically elegant relational algebra.

This restriction creates impedance mismatch. Meaning, the choice of using algebraic operations introduces limitations in flexibility, such as rows unable to contain certain data structures (e.g. lists and sets). To retrieve a collection of values in a relational database, we need to perform joins and apply complicated operations to tuples. In other words, algebraic operations provide simplicity and elegance, but also creates limitations.

Most importantly, relational databases can produce limitations in usable resources. At a high level, increasing the growth of data either requires scaling upwards or outwards. Needless to say, scaling upwards has an actual cut-off point, since a machine can only grow so large.

On the other hand, scaling outwards is possible, depending on the type of database. For relational databases, scaling outwards can become difficult in the long run, since relational databases aren't built for scaling outwards. Specifically, they aren't built to run on clusters. Methods that feel like workarounds, like sharding, can be performed, but these can become very difficult to manage in the long run.

What is NoSQL

The notion of NoSQL databases has been around since the 90s. Since then, databases have come a long way and almost seem unrecognizable. Thus, NoSQL databases were more easily distinguishable when they were first introduced, but the line separating SQL and NoSQL databases has blurred.

Today, the term NoSQL has mostly become a buzzword, since it is harder to classify since its origin. Obviously, NoSQL databases are mostly driven by the absence of SQL. To illustrate my previous point regarding a more blurred line, a NoSQL database like Cassandra has CQL. Although CQL is far from supporting the flexibility of standard SQL, a basic querying language still is offered in Cassandra nonetheless.

Even today, most NoSQL databases are schemaless. Meaning, they're given the flexibility and freedom to specify data values without types in most cases. As stated earlier, NoSQL databases are mostly driven by the need to run on clusters. On the other hand, relational databases favor consistency and availability, rather than partition-tolerance.

Purpose of NoSQL Databases

As hinted at previously, the clearest benefits of NoSQL databases are scalability and productivity related to application development. In most cases, developers use in-memory data structures that aren't naturally designed as a relational data model. If these developers want to store their data in a relational database, they'll need to map their in-memory data structures to a relational data model.

Too much effort on application development can be spent during this step. This mapping process can be eliminated entirely by introducing a NoSQL database offering a more comparable data model. Making this replacement simplifies any interaction between the application and database, which results in less code to write, debug, and change over time. In summary, developers can more easily store their in-memory data structures in NoSQL databases, since many of them don't enforce as strict layouts.

In comparison to relational database, NoSQL databases are built for scaling large data. For example, the architecture of Cassandra was designed around concepts such as consistent hashing, partitioning, and replication. Furthermore, quickly capturing increased amounts of data can be more expensive with relational databases, since relational databases are designed to run on a single machine. Indicating, the additional hardware for being able to hold this data becomes quite expensive.

On the other hand, a more economical alternative involves storing and computing large amounts of data on large clusters of smaller and cheaper machines. Since the majority of NoSQL databases explicitly are designed to run on clusters, they become the preferred option.

Properties of NoSQL Databases

Each NoSQL database has its own unique set of properties. However, they generally share a few high-level properties with each other. As mentioned earlier, NoSQL databases don't support ACID transactions. The exception to this is a graph database. To use another example from earlier, NoSQL databases introduce flexibility from being schemaless.

To go into greater detail, schemas are defined in the inital stages of a relational database. Whereas, they are not defined in NoSQL databases. By doing this, NoSQL sacrifices data integrity for increased flexibility and speed of development. Schemaless databases are generally useful when we're not entirely certain about the structure of the data being stored, which happens often in the early stages of a project.

Schemaless databases also are generally useful when we're dealing with nonuniform data. Meaning, certain rows may be associated with their own set of columns. Potentially, this can happen when storing a sparse matrix. Lastly, schemaless databases are generally useful when there isn't enough time (or a high enough priority) to think about the exact structure of data. Specifically, this may occur when creating adhoc reports.

Without using a strict schema, any data can be stored under a key in a key-value model. A document model essentially emulates this approach, since it doesn't enforce any additional (major) restrictions on the document. Column-families also ensures any data can be stored in its model, but forces data values to be stored under a column-family. Alternatively, graph databases store any data as nodes and edges with properties.

There is also a greater range of relationships offered by NoSQL databases, compared to relational databases. Most of these relationships are better suited for application developers.

When choosing one NoSQL database over the other, a developer needs to recognize the distinctive set of requirements for his or her application. Some of these requirements will align with a key-value database, and others will align with a graph database. Generally, the type of application can guide the developer to a NoSQL database. For example, recommendation systems are better suited for graph databases, whereas session caching is better suited for key-value databases.

Lastly, most NoSQL database offer some form of materialized views, which are supported in relational databases as virtual tables. They aren't handled the same way as views in relational databases, but they essentially achieve the same goal. In NoSQL databases, materialized views refer to precomputed and cached queries. These queries are typically assembled using a Map-Reduce or Spark job, since many of these NoSQL databases don't support a querying langauge.

Introducing Data Models in Databases

Regardless of whether a database is relational or non-relational, each database supports a data model. A data model refers to the low-level structure of data being stored in a database. The structure of a data model impacts how we interact with the data in a database.

A relational databases is known to use a relational data model. As hinted at earlier, a relational data model consists of relations and tuples. Again, relations represent tables, and tuples represent rows intuitively. Indicating, a relational model stores its data as tuples.

In comparison to NoSQL data models, tuples are rather inflexible data structures, since one tuple can't be nested within another tuple. To retrieve nested records, tuples expect algebraic operation to be performed.

At a high level, a relational model doesn't allow lists of values to be stored in a field, since tuples can't be nested. This restriction is the reason for the relational model's inflexibility. For example, application developers have a difficult time storing their in-memory data structures as a relational model, since they need to go through extra effort to convert their data structures.

Alternatively, aggregates support nesting, meaning they can store lists of values. In the book NoSQL Distilled, the term aggregate is used to refer to:

A collection of related objects that we wish to treat as a unit. In particular, it is a unit for data manipulation and management of consistency.

Aggregate-oriented databases have their obvious consequences. In general, aggregate-oriented databases don’t have ACID transactions across multiple aggregates. Instead, each aggregate is manipulated individually. Meaning, atomic manipulation of multiple aggregates must be handled in the actual application code.

Types of Aggregate Data Models

After introducing the overlapping properties of NoSQL databases, we'll now explore the specifics of certain types of NoSQL databases. Various forms of NoSQL databases were briefly mentioned in the earlier segments of this post, which include key-value, document, column-family, and graph databases.

A key-value and document are strongly aggregate-oriented databases, meaning they consist of many aggregates. In particular, key-value databases generally enforce fewer restrictions and offer increased flexibility, compared to both relational databases and other NoSQL databases. In certain situations, there are very minor constraints, such as size limits. However, they generally offer more freedom comparatively.

In key-value databases, aggregates are only accessible by means of a key. As a result, values can't be queried and requires a lookup for the entire aggregate.

Document databases generally enforce similar constraints as key-value databases. However, they can sometimes enforce a few additional limitations, such as:

Size limits
What we can place in them
Data types

Document databases remain schemaless. Additionally, the fields in document databases can be queried, which contrasts with key-value databases. In other words, we can query specific segments of an aggregate in document databases, whereas we can only query the entire aggregate (belonging to a key) in key-value databases. Lastly, indices can be created based on the contents of an aggregate in a document database.

As mentioned previously, there are workarounds that blur the line between document and key-value databases. For example, certain key-value databases, such as Redis, can hack together something that feels like indexing by separating aggregates into lists. However, we generally expect to access aggregates using a key in a key-value database. Whereas, we generally expect to access aggregates using some form of querying based on values in a document database. Keep in mind, this query may or may not include a key.

Column-family databases impose even more limitations on the structure of an aggregate. Specifically, column-family databases organize their columns into column families, which was mentioned earlier as well. Each column needs to be assigned to a particular column family. Then, each column can be accessed via a column family. In particular, accessing a column-family will return each of the columns associated with that column family.

The above details about NoSql databases only scratch the surface of their structure and behavior. For a more detailed explanation about key-value databases, refer to this post. For a more detailed explanation about document databases, refer to this post. Then, refer to this post for details about column-family databases. Finally, for a deeper dive into graph databases, refer to this post.

Use Cases of Database Types

The table below consists of a few types of databases, along with their use cases. For even more use cases, refer to this AWS article.

Database Type	Use Cases	Database
Relational	Traditional applications; basics relationships; general reporting	MySQL; PostgreSQL
Key-Value	High-traffic applications; storing configurations; session caching	Redis; Riak
Document	Content management; user profiles; prototypes; big data analytics	MongoDB; Cassandra
Graph	Complex relationships; fraud detection; recommendation engines	Neo4j

For a more detailed description about the differences between column-oriented databases and column-family databases, refer to this post. For a useful point about the behavior of join operations in NoSQL databases, refer to this post. For a more detailed comparison between NoSQL databases, refer to this article. For informative presentations about NoSQL databases, refer to this video and this video.

File System and Database Differences

Sun, 02 Feb 2020 00:00:00 GMT

In most cases, database storage is implemented using file system files, where databases are usually stored in files, which exist in filesystems. The data within a database are usually stored in files, which can be found in a directory structure and generally follow naming conventions assigned by the manufacturer of the DBMS.

DBMS and File Systems

The major differences between a database and a file system are outlined in this useful post. First, database engines organize data into files containing records as their data model. These records are obviously represented as byte sequences at the lowest level, but abstracting these sequences as records allows the DBMS to include additional querying functionality. On the other hand, file systems organize data into files and manage them as byte sequences, leading to less querying functionality since there is little awareness of the file contents beyond the byte level.

Not only does managing records introduce additional querying functionality, but it also introduces search functionality in most cases. Since databases create a layer of abstraction by using records as its data model, they can retrieve and modify those records more easily using search capabilities managed by the database engine. For example, a DBMS may allow users to search for records via SQL or key-value pairs, whereas most basic file systems don't come equipped with as detailed search functionality.

In the end, storing data is just storing data. A file just represents some bytes at the lowest level, but it is a storage abstraction. A database is a storage abstraction, and a file system is a storage abstraction. Databases ultimately store their data as files behind the scenes, whereas file systems store data as files too. Databases and basic file systems usually differ by how much additional functionality is offered, where databases typically offer more querying and search functionalities. More details about the differences in functionalities can be found in this post.

File Storage and Object Storage

Both file and object storage are different types of storage used for saving and managing data as files with metadata. For example, HDFS is a distributed file storage and can use Hive for storing metadata, and most object stores maintain metadata on their objects as well. Traditional object storage typically stores data as binary files and uses HTTP for accessing the data, whereas file stores hold data formatted in many different file types and can be accessed using SFTP in most cases. For specific examples of open-source object stores (that aren't AWS S3), refer to this repository on GitHub.

Most databases today are file storage by default, where some databases can be enabled to be block storage. Again, most databases are file stores with some extra functionality, such as querying capabilities, search functionality, tracking more extensive metadata, inclusion of a GUI, etc. Relational and NoSQL databases physically store data in a semi-similar fashion, meaning they both chop up data into partitions and store these partitions as binary files. They really just differ on how they abstract data, or how they locally store data into logical data models internally.

Relational databases, such as SQL server, partition data files into $8K$ binary files with some metadata by default. An RDBMS logically stores data into tables. On the other hand, file system storage, such as HDFS, partitions data files into $128MB$ binary files with some metadata by default. A file system logically stores data into files in a fairly standard, hierarchical file system with directories in most cases. For HDFS, the Hive add-on can provide a SQL-like interface by creating additional metadata for the files. Technically, HDFS isn't considered a database, since it's a distributed file system.

One example of a non-relational database is a key-value database. MongoDB is a common key-value database that partitions data files into BSON (or binarized JSON files) with some metadata. In this case, a key-value non-relational database logically stores data as key-value pairs in physical binary files. Another example of a non-relational database is a columnar store. Cassandra is a common columnar database that partitions data files into binary files with some metadata. In this case, a columnar non-relational database logically stores data as tables. Other examples of columnar databases include Google's Bigtable, Snowflake, and others.

Database Sharding

Mon, 06 Jan 2020 00:00:00 GMT

Query optimization, indexing, and NoSQL solutions are all popular scalability strategies when designing server-side systems. If those options aren't enough, then sharding may be the next best strategy for optimizing or scaling a monolithic RDBMS. For more information about techniques used in indexing, refer to the previous post.

Sharding refers to the process of separating a large table into smaller subsets, which are spread across multiple servers. These smaller chunks are called shards.

In particular, sharding is commonly used for horizontally scaling a monolithic RDBMS. Implying, a shard is a horizontal data partition, where each partition contains a subset of the larger table. When the data within a table grows too large for a single server, we can use sharding to store subsets of data on multiple nodes.

Other reasons for implementing sharding relates to limitations in memory capacity and compute power. When data grows very large in an unsharded database, any maintenance and query performance becomes slow. Note, vertical scaling has its own limitations in reference to adding resources to support database operations.

As stated previously, shards are stored on database nodes within a cluster. A MySQL cluster automatically partitions tables (or shards) across nodes. By doing this, MySQL enables the database to scale horizontally on cheap, commodity hardware.

With the improvements of sharding mentioned previously, we can generally say that sharding improves the scalability and availability of an RDBMS. Specifically, separating a much larger table into smaller shards will increase the amount of compute capacity for serving incoming queries. Therefore, we'll end up with faster query response times and index builds. Similarly, creating a strategy involving horizontal scaling increases the storage capacity within the database. With respect to cost, a network of smaller and cheaper servers may be more cost effective in the long term than maintaining one big server.

Additionally, sharding can offer increased availability. During downtime, the data within an unsharded database is inaccessible. In a sharded database, nodes experiencing downtime will only ensure downtime for its shards. Meaning, any nodes that remain online will be available for read and write operations.

Essentially, sharding can be implemented by taking partitions and storing them on separate nodes. Then, we need to think about how to shard our data, which is usually performed on a column. In most cases, we shard a table on the primary key. For example, we could shard a dataset on a $\text{userid}$ or $\text{locid}$ .

Typically, optimizing join operations is a commonly experienced problem when separating our data into shards. Specifically, queries will perform poorly when performing join operations, since the join needs to gather data from different shards across the network, which is costly.

In addition, there needs to be a flexible way for adding and removing shards within a cluster. Consistent hashing can help solve this problem, which is mentioned in the previous post. Specifically, consistent hashing can be used for indexing data into shards. Alternatively, a hierarchical sharding protocol also can help reduce poor performance of this problem. Lastly, we can optimize the performance of queries on shards by by creating an index on each shard.

For more a more detailed explanation about sharding, refer to this video or this article. For a more detailed analysis of sharding techniques, refer to this post.

Consistent Hashing for Load Balancing

Fri, 03 Jan 2020 00:00:00 GMT

As you know, a hash function maps key values to index values. Typically, these functions are used to determine the location (i.e. index) of a record within a table. They have other applications, such as mapping an image to some hash value. Hash functions are used on a much broader scale for password verification and detection of changes to data. The following image illustrates a few use cases for hash functions.

A hash function represents the binary code representing any hashed meidum. For example, a song can be represented as binary code, which can be input into a hash function. Similarly, an image can be represented as binary code, which can be input into a hash function as well. Likewise, string, integers, etc. can all be represented as binary code, which is the input of a hash function.

In terms of load balancing, a load balancer receives a request from a client and returns a response from a designated server on our network. If we make multiple servers available to the client, then the load balancer needs to determine which server will provide the client with the fastest possible response time. Typically, this process involves equally distributing the number of requests to the servers within our network. The traditional method for solving this problem involves the use of the modulo operator. For a more detailed explanation about basic hashing, refer to this video.

Consistent hashing can be found in caching throughout load balancing, such as Uber's open-sourced RingPop project. Broadly speaking, consistent hashing maps objects to the same cache machine as far as possible. When a machine used for caching is added to the network, the machine takes its share of objects from all the other cache machines and when it is removed, its objects are shared among the remaining machines.

The primary notion behind consistent hashing involves associating each cache with one or more has value intervals, whether the interval boundaries are determined by calculating the hash of each cache identifier. Once the cache is removed, its interval is taken over by a cache with an adjacent interval, while all the remaining caches are unchanged.

More specifically, this process can be defined using the illustration below, and in these illustrations. First, we implement a hash ring containing an $M-1$ amount of hashed request ids. Then, we map the available servers in our network on the ring uniformly. Each server is associated with its own hashed request id, which represents a boundary. Once a load balancer receives a request, it maps the hashed request onto the hash ring. Then, this hashed request is mapped to a server immediately clockwise to it.

Since a request is mapped to an immediate server in the clockwise direction, the addition or removal of a server will uniformly affect fewer requests in theory. In other words, we can expect the load to be equally distributed on average, since the hash values are assigned randomly and uniformly.

In practice, there is a greater change of having non-uniform distribution of requests between servers, since we often have a small number of servers on our hash ring. The image below illustrates a scenario where nodes are removed from our ring, creating a non-uniform distribution.

To prevent non-uniform distribution from occurring in our hash ring, we can introduce the idea of virtual nodes. Virtual nodes refer to multiple instances of a server on the hash ring. By using a $k$ number of hash function, each server corresponds to $k$ number of hash values on the ring. This increases the randomness of the load. For a more detailed understanding of consistent hashing and its use cases, refer to this video and this article by Tom White.

Consistent hashing has a few advantages over standard hashing. In particular, it involves a minimal amount of data transfer between machines, which has been proven to be the most optimal amount. However, there are still a few areas of improvments, such as the client needing to know the number of nodes. The client also must know the location of each node on the circle. To improve upon the method mentioned above, we can implement sharding techniques, which are mentioned here.

The table below illustrates the asymptotic time complexities for $N$ nodes and $K$ keys. The $O(K/N)$ complexity refers to the average cost for redistribution of keys, whereas the $O(logN)$ complexity occurs from a binary search among nodes, in order to find the next node on the ring.

	Classic Hashing	Consistent Hashing
Adding a node	$O(K)$	$O(K/N + logN)$
Removing a node	$O(K)$	$O(K/N + logN)$
Adding a key	$O(1)$	$O(logN)$
Removing a key	$O(1)$	$O(logN)$

Consistent Hashing isn't only used in load balancing. It can also be found in database indexing. Generally, there are a few ways to improve the performance of an RDBMS database system. First, we may be interested in very particular performance improvements, so optimizing queries may be the extent of the problem. For broader improvements to the performance of a database, we may want to perform indexing on certain tables.

For database indexing, a B-tree index is used for column comparisons in expressions that use the $>$ and $\le$ operators. A B-tree index can also be used for LIKE comparisons. On the other hand, hash indexes are used for equality comparisons, which involve the $=$ and $<>$ operators. For more information about the use cases of B-tree indexing and hash indexing, refer to the MySQL docs.

There are a few other improvements we could implement. For example, we may be interested in translating our SQL database over to a NoSQL database, depending on the structure of data being saved. In most cases, it will not be worthwhile to make this change. Lastly, we can implement a data sharding strategy, which is discussed more in the next post.

Scaling a Pizza Chain

Wed, 01 Jan 2020 00:00:00 GMT

In computer science, systems design refers to the process of defining and developing a system that satisfies certain requirements made by the user. Obviously, this involves a detailed understanding of the components within a system and how they interact with each other. In particular, this process usually involves knowing the input requirements, output requirements, storage requirements, processing requirements, etc. Specifically, the following concepts can help improve a particular system being managed:

Vertical scaling
Scheduling batch jobs
Assigning backup servers
Horizontal scaling
Developing a distributed system
Load balancing
Logging and calculating metrics

Interestingly, these same concepts can be applied to many problems outside of computer science, such as running a pizza chain efficiently. This post was motivated by this great video, so please refer to it for more details.

Running a Single Pizza Shop

In this example, we'll consider a single pizza parlor as a cluster, whereas a single chef is a single computer. Then, we can think of an order as a request to our server. In the technical sense, horizontal scaling would imply hiring additional chefs to our pizza parlor. On the other hand, vertical scaling would imply purchasing additional technology for an individual chef to help them work harder.

In a distributed system, our goal is to increase the throughput of incoming requests by optimizing resources and running processes. One way of achieving this in our pizza shop analogy is through vertical scaling, or expanding the limits of a chef so they can handle more work. This improvement to the system is analogous to the change to the pizza shop illustrated below.

Preparing Pizzas in Advance

In most cases, the peak hours of a pizza shop is during the day. In order to ease the workload on our chef during these peak hours, he or she can prepare pizzas during the night. The chef would prefer preparing pizzas after hours to avoid preparing pizzas during the day, when he or she has many orders to handle.

In computer science, this could look like setting up batch jobs to process large amounts of data during the nights. Usually, these could be in the form of cron jobs, and they can run for a few hours during 3AM-4AM. Specifically, we wouldn't want to run these batch jobs during the day, since this could harm the performance of users querying our database.

Recovering from Failures Quickly

In any system, we want to avoid having a single point of failure. Indicating, our system should be resilient. When managing a pizza store with a single chef, there should be one or more chefs on call, in case the chef calls in sick. Otherwise, the business would be shut down for the day the chef calls in sick, causing this to be a single point of failure. In our system, we would want to create backup servers in this case. Then, our backup server would fill in for our worker server.

Hiring Additional Chefs

Clearly, only employing a single chef at our pizza shop becomes a greater problem if we need to handle a much larger number of pizza orders. Hiring additional chefs can also prevent a single point of failure, since our shop still can function if one chef calls off. In other words, we can hire more pizza chefs to scale up our pizza shop.

In a distributed system, this concept translates to purchasing additional servers to handle more requests. This improvement to the system is analogous to the change to the pizza shop illustrated below.

Vertical scaling and horizontal scaling typically are used together in a system. When systems observe a problem requiring either horizontal scaling and vertical scaling, we can make the appropriate changes based on the properties of the problem. In particular, we can use the more general use cases provided below to determine which type of scaling should be applied.

Property	Horizontal	Vertical
Load Balancing	Yes	N/A
Resilient	Yes	Single point of failure
Speed of Communication	Slower (RPC)	Faster (IPC)
Consistency	Inconsistent	Consistent
Scalability	Scales well as users increase	Hardware limits

Adding Additional Pizza Shops

For creating a truly scalable pizza chain, there should be more than one pizza shop. To illustrate this point, suppose our shop experiences a power outage for an entire day. In this case, our pizza shop couldn't accept any orders and would need to close down for the day. Expanding our pizza shop to other locations could help solve this problem.

By building additional shops, we still can handle orders in case a single shop experiences a power outage. Also, pizzas can be delivered at a faster rate if orders are handled at a shop closter to the customer. These pizza shops need to be able to communicate with each other, but should be able to handle orders on their own in case one shop isn't available.

Adding additional shops is analogous to creating new servers or even clusters in a distributed system. Similarly, these servers need to be able to communicate with each other. Creating new servers or clusters leads to a more fault-tolerant and responsive distributed system. For example, AWS will include multiple servers from different datacenters within their clusters, which leads to faster response times.

Efficiently Managing Pizza Orders

Ideally, customers would submit an order for delivery to an entry point for our delivery services, which could be a site, app, or specific number. Then, a manager would determine a store that efficiently prepares the pizza and delivers it in the fastest possible time frame. The image below illustrates a load balancer as a central office for our delivery services.

In a distributed system, the component responsible for managing the load and forwarding of a request is known as a load balancer. Specifically, a load balancer is responsible for dispatching requests based on statistics saved by it or some other external agent. In other words, a load balancer will receive a request and determine the optimal node that should process the request.

Monitoring the Performance of our Shops

As we build even more shops and hire additional chefs to handle the demand for our delicious pizza, monitoring the performance, decisions, and actions of our employees and shops becomes difficult. As a result, logging the decisions and calculating metrics on these components becomes critical. Automating behavior based on these logs becomes even more critical if we want to build a truly scalable pizza shop.

Similarly, we should log information, warning, and errors within our system to measure. In other words, creating a system for logging is important to ensure nodes and clusters are behaving as expected.

Testing Spark Applications with Mesos

Thu, 04 Jul 2019 00:00:00 GMT

This post walks through an example of running a cluster using a Mesos cluster manager on Mac OS. In the coming posts, we'll explore other examples, including clusters running a standalone cluster manager and a cluster manager in YARN.

Describing the Mesos Architecutre
Comparing Mesos and Standalone Architectures
Setting up Mesos
Setting up a SparkSession
Launching Mesos Daemons
Launching Spark Daemons
Accessing Web UI for Daemons
Launching Applications in Client Mode

Describing the Mesos Architecture

The Mesos architecture is arguably more similar to the standalone architecture, compared to the YARN architecture. This is because the essential components of the Mesos architecture include a master and workers only. In the Mesos architecture, a master schedules worker resources for applications that need to use them. Then, workers launch executors for applications, which execute tasks.

Mesos can run Docker containers. As a result, Mesos essentially can run any application that can be set up in a Docker container, which includes Spark applications.

Generally, Mesos is more powerful than a cluster with a Spark standalone cluster manager. Mesos can be used for applications other than Spark, as well. Specifically, it can be used for Java, Scala, Python, and other applications. It can also do more than schedule CPU and RAM resources. In particular, it is capable of scheduling disk space, network ports, etc.

Spark applications running on Mesos consist of two components. These two components include a scheduler and executor. The scheduler accepts or rejects CPU and RAM resources offered by the Mesos master. Then, this master automatically starts Mesos workers, which automatically start executors. Lastly, Mesos executors run tasks as requested by the scheduler.

Comparing Mesos and Standalone Architectures

Resource scheduling is the most distinct difference between the Mesos and standalone architectures. Specifically, a standalone cluster manager automatically assigns resources to applications. A Mesos cluster manager optionally offers resources to applications. In this case, an application can accept and refuse the resources.

Setting up Mesos

Install Mesos
```
$ brew install mesos
```
Install the ZooKeeper dependency
```
$ brew install zookeeper
```
Optionally, assign IP addresses for masters in /etc/mesos/zk:
```
zk://192.0.2.1:2181,192.0.2.2:2181/mesos
```

Setting up a SparkSession

Download Spark 2.4.6
Add the path for Spark in .bash_profile:

export SPARK_HOME=./spark-2.4.6-bin-hadoop2.7

Create the file ./conf/spark-defaults.conf:

spark.master=yarn
spark.driver.am.memory=512m
spark.yarn.am.memory=512m
spark.executor.memory=512m
spark.eventLog.enabled=true
spark.eventLog.dir=./tmp/spark-events/
spark.history.fs.logDirectory=./tmp/spark-events/
spark.driver.memory=5g

Create a Spark application:

# test.py
>>> from pyspark import SparkContext
>>> file = "~/data.txt"  # path of data
>>> masterurl = 'spark://localhost:7077'
>>> sc = SparkContext(masterurl, 'myapp')
>>> data = sc.textFile(file).cache()
>>> num_a = data.filter(lambda s: 'a' in s).count()
>>> print(num_a)
>>> sc.stop()

Launching Mesos Daemons

Start the master:
```
$ sudo service mesos-master start
```
Start a worker:
```
$ sudo service mesos-slave start
```
Stop the master:
```
$ sudo service mesos-master stop
```
Stop a worker:
```
$ sudo service mesos-slave stop
```

Launching Spark Daemons

Start a master daemon in standalone mode
```
$ ./sbin/start-master.sh
```

Start a worker daemon

$ ./sbin/start-slave.sh spark://localhost:7077

Start a history daemon
```
$ ./sbin/start-history-server.sh
```

Start a Spark application

$ ./bin/spark-submit \
--master mesos://localhost:5050 \
test.py

Stop the daemons

$ ./sbin/stop-master.sh
$ ./sbin/stop-slave.sh
$ ./sbin/stop-history-server.sh

Accessing Web UI for Daemons

Mesos provides a web UI for each initialized daemon. By default, Spark creates a web UI for the master on port 5050. The workers can take on different ports and can be accessed via the master web UI. The history server can be accessed on port 18080 by default. The table below summarizes the default locations for each web UI.

Daemon	Port
Mesos Master	`5050`
Spark History	`18080`

Launching Applications in Client Mode

Mesos workers offer their resources to the master
The Mesos scheduler registers with the Mesos master
- The Mesos master is located in the cluster
- The Mesos scheduler is Spark's Mesos-specific scheduler
- The Mesos scheduler runs in the driver
The Mesos master offers available resources to the Mesos scheduler
- This happens continuously
- The offer is sent out every second while master is alive
The Mesos scheduler accepts some of the resources
The Mesos scheduler sends metadata about the resources to the Mesos master
- This metadata includes information about these resources and tasks that run these resources
The Mesos master asks the workers to start the tasks with its specified resources
The Mesos workers launch Mesos executors
The Mesos executors launch Spark executors consisting of Tasks
The Spark executors communicate with the Spark driver

Testing Spark Applications with YARN

Mon, 17 Jun 2019 00:00:00 GMT

This post walks through an example of running a cluster using a YARN cluster manager on Mac OS. In the coming posts, we'll explore other examples, including clusters running a standalone cluster manager and Mesos cluster manager.

Describing the YARN Architecutre
Comparing YARN and Standalone Architecture
Setting up Hadoop
Setting up a SparkSession
Launching YARN Daemons
Launching Spark Daemons
Accessing Web UI for Daemons
Launching Applications in Client Mode
Launching Applications in Cluster Mode

Describing the YARN Architecture

There are many redundancies found throughout the standard YARN architecture compared to the standalone architecture in Spark. There are a few additional components in YARN that replaces some of the daemons in the standalone architecture:

Resource Manager
Node Manager
Containers
Application Master

Essentially, the resource manager is the same as the master process in Spark's standalone mode. The node manager is essentially the same as the worker process. There is a single resource manager per cluster and a single node manager per node in the cluster.

Rather than representing executors and processes as JVM instances, YARN represents them as containers. However, each containers is still run as a JVM with a requested heap size. These containers contain an application master, which is responsible for requesting application resources from the resource manager.

When an application is run using YARN, the driver process acts as the YARN application master in Spark. Then, node managers monitor CPU and RAM resources used by containers. As a result, they report these resources to the resource manager.

Comparing YARN and Standalone Architectures

Although a the standalone Spark cluster manager and the YARN cluster manager has a lot of similarities, some of the responsibilities change and JVM instances behave differently. Primarily, resource scheduling is performed by the master JVM in standalone mode, whereas it is performed by the resource manager in YARN.

Executors are asked to start by the master JVM in standalone mode, whereas they are asked to start by the application master in YARN. Job scheduling still is performed by the Spark scheduler in both modes. When Spark is running on YARN, the Spark driver process acts as the YARN application master. Additionally, YARN refers to its processes as containers, rather than JVM instances in standalone mode.

Setting up Hadoop

Install Hadoop:

$ brew install hadoop

Download Java version supported by Hadoop 3.0:

$ brew cask install java8

Configure the path of Java ran by Hadoop:

# /usr/local/Cellar/hadoop/3.2.1_1/libexec/etc/hadoop-env.sh
export JAVA_HOME=/Library/Java/JavaVirtualMachines/jdk1.8.0_231.jdk/Contents/Home

Configure the HDFS address:

<!--/usr/local/Cellar/hadoop/3.2.1_1/libexec/etc/core-site.xml-->
<configuration>
  <property>
    <name>
      hadoop.tmp.dir
    </name>
    <value>
      /usr/local/Cellar/hadoop/hdfs/tmp
    </value>
    <description>
      A base for other temporary directories
    </description>             
  </property>
  <property>
    <name>
      fs.default.name
    </name>
    <value>
      hdfs://localhost:8020
    </value>
  </property>
</configuration>

Configure the MapReduce JobTracker address:

<!--/usr/local/Cellar/hadoop/3.2.1_1/libexec/etc/mapred-site.xml-->
<configuration>
  <property>
    <name>
      mapred.job.tracker
    </name>
    <value>
      localhost:8021
    </value>
  </property>
</configuration>

Configure the HDFS properties:

<!--/usr/local/Cellar/hadoop/3.2.1_1/libexec/etc/hdfs-site.xml-->
<configuration>
  <property>
    <name>
      dfs.replication
    </name>
    <value>
      1
    </value>
  </property>
</configuration>

Configure SHH Keys:

$ ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa
$ cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys
$ chmod 0600 ~/.ssh/authorized_keys

Setting up a SparkSession

Download Spark 2.4.6
Add the path for Spark in .bash_profile:

export SPARK_HOME=./spark-2.4.6-bin-hadoop2.7

Create the file ./conf/spark-defaults.conf:

spark.master=yarn
spark.driver.am.memory=512m
spark.yarn.am.memory=512m
spark.executor.memory=512m
spark.eventLog.enabled=true
spark.eventLog.dir=./tmp/spark-events/
spark.history.fs.logDirectory=./tmp/spark-events/
spark.driver.memory=5g

Create a Spark application:

# test.py
>>> from pyspark import SparkContext
>>> file = "~/data.txt"  # path of data
>>> masterurl = 'spark://localhost:7077'
>>> sc = SparkContext(masterurl, 'myapp')
>>> data = sc.textFile(file).cache()
>>> num_a = data.filter(lambda s: 'a' in s).count()
>>> print(num_a)
>>> sc.stop()

Launching YARN Daemons

Start YARN from the NameNode:
```
$ ./bin/start-yarn.sh
```

The YARN cluster manager needs to be started on the NameNode. By doing this, the ResourceManager and NodeManagers should be started using the command above. Specifically, we should see the following lines after running the command above.

Starting resourcemanager
Starting nodemanagers

Stop the daemons
```
$ ./bin/stop-yarn.sh
```

Launching Spark Daemons

Start a master daemon in standalone mode
```
$ ./sbin/start-master.sh
```

Start a worker daemon

$ ./sbin/start-slave.sh spark://localhost:7077

Start a history daemon
```
$ ./sbin/start-history-server.sh
```

Start a Spark application

$ ./bin/spark-submit \
--master yarn \
test.py

Stop the daemons

$ ./sbin/stop-master.sh
$ ./sbin/stop-slave.sh
$ ./sbin/stop-history-server.sh

Accessing Web UI for Daemons

Spark provides a web UI for each initialized daemon. By default, Spark creates a web UI for the master on port 8080. The workers can take on different portsand can be accessed via the master web UI. The history server can be accessed on port 18080 by default. The table below summarizes the default locations for each web UI.

Daemon	Port
YARN Master	`8080`
YARN Worker	`8081`
Spark History	`18080`
HDFS Resource Manager	`9870`
YARN JobTracker	`8088`

Launching Applications in Client Mode

Client's JVM process submits a driver to the resource manager
The driver is launched
The resource manager instructs a node manager to start a container with an application master
- The container includes the application master
- The resource manager represents the master
- The node manager represents the worker
- The application master requests for resources
Node manager launches a container with an application master
Application master requests the resource manager to allocate resources for the application
App master asks node managers to start executor containers
Node managers launch executors
- This is on behalf of the Spark application master
The driver and executors communicate independently
- Doesn't involves the master or workers

Launching Applications in Cluster Mode

Client's JVM process submits a driver to the resource manager
The resource manager instructs a node manager to start a container with an application master
- The container includes the application master
- The resource manager represents the master
- The node manager represents the worker
- The application master requests for resources
Node manager launches a container with an application master
- The application master contains the spark driver
Application master requests the resource manager to allocate resources for the application
App master asks node managers to start executor containers
Node managers launch executors
- This is on behalf of the Spark application master
The driver and executors communicate independently
- Doesn't involves the master or workers

Testing Spark Applications in Standalone

Thu, 06 Jun 2019 00:00:00 GMT

This post walks through an example of a cluster running in standalone mode. In the coming posts, we'll explore other examples, including clusters running a YARN cluster manager and Mesos cluster manager.

Setting up a SparkSession
Launching Daemons
Accessing Web UI for Daemons
Caveat about PySpark Applications
Launching Applications in Client Mode
Launching Applications in Cluster Mode

Setting up a SparkSession

Download Spark 2.4.6
Create the file ./conf/spark-defaults.conf:

spark.master=spark://localhost:7077
spark.eventLog.enabled=true
spark.eventLog.dir=./tmp/spark-events/
spark.history.fs.logDirectory=./tmp/spark-events/
spark.driver.memory=5g

Create a Spark application:

# test.py
>>> from pyspark import SparkContext
>>> file = "~/data.txt"  # path of data
>>> masterurl = 'spark://localhost:7077'
>>> sc = SparkContext(masterurl, 'myapp')
>>> data = sc.textFile(file).cache()
>>> num_a = data.filter(lambda s: 'a' in s).count()
>>> print(num_a)
>>> sc.stop()

Launching Daemons

Start a master daemon in standalone mode
```
$ ./sbin/start-master.sh 
```

Start a worker daemon

$ ./sbin/start-slave.sh spark://localhost:7077

Start a history daemon
```
$ ./sbin/start-history-server.sh
```

Start a Spark application

$ ./bin/spark-submit \
--master spark://localhost:7077 \
test.py

Stopping the daemons

$ ./sbin/stop-master.sh
$ ./sbin/stop-slave.sh
$ ./sbin/stop-history-server.sh

Accessing Web UI for Daemons

Daemon	Port
Master	`8080`
Worker	`8081`
History	`18080`

Caveat about PySpark Applications

Notice, launching an application in client mode doesn't seem to trigger a driver according to the master's web UI. This doesn't mean a driver isn't launched in client mode. The driver is still launched within the spark-submit process. However, the master's web UI omits driver information if the application is running in client mode.

So, we may want to launch an application in cluster mode now. However, running an application in cluster mode would give us the following error:

$ ./bin/spark-submit \
    --master spark://localhost:7077 \
    --deploy-mode cluster
    test.py
Exception in thread "main" org.apache.spark.SparkException: Cluster deploy mode is currently not supported for python applications on standalone clusters.

As of Spark 2.4.6, we can't run python applications in cluster mode when running a standalone cluster manager. This is a good opportunity for us to experiment with other resource managers in the next post. For now, we will run JavaSparkPi.java found in the examples directory.

Launching Applications in Client Mode

In a previous post, we defined the components associated with a driver program and cluster, while illustrating the interaction between a driver program and cluster components. Specifically, we defined this interaction when applications are launched in client mode. Now, we can execute an application and verify these steps using the logs.

Note, the timestamps and logged messages were slightly modified for clarification. However, the order and substance of each message still remains the same.

14:43:01 INFO
SparkContext: Submitted application: Spark Pi

14:43:02 INFO
Utils: Successfully started service 'sparkDriver'

14:43:03 INFO
StandaloneAppClient: Connecting to master

14:43:04 INFO
StandaloneSchedulerBackend: Connected to Spark cluster

14:43:05 INFO
Master: Registered app Spark Pi

14:43:06 INFO
Master: Launching executor on worker

14:43:07 INFO
Worker: Asked to launch executor

14:43:08 INFO
ExecutorRunner: Launched

14:43:09 INFO
StandaloneAppClient: Executor added on worker

14:43:10 INFO
StandaloneSchedulerBackend: Granted executor ID

14:43:11 INFO
StandaloneAppClient: Executor is now RUNNING

14:43:12 INFO
SparkContext: Starting job

...

14:43:13 INFO
DAGScheduler: Job finished

14:43:14 INFO
StandaloneSchedulerBackend: Shutting down all executors

14:43:15 INFO
Worker: Asked to kill executor

14:43:16 INFO
ExecutorRunner: Killing process!

14:43:17 INFO
Master: Removing app

14:43:18 INFO
SparkContext: Successfully stopped SparkContext

Launching Applications in Cluster Mode

In a previous post, we defined the interaction between a driver program and cluster components, while applications are launched in cluster mode. Now, we can execute an application in cluster mode to verify these steps using the logs.

Note, the timestamps and logged messages were slightly modified for clarification. However, the order and substance of each message still remains the same.

14:43:01 INFO
Master: Driver submitted

14:43:02 INFO
Master: Launching driver

14:43:03 INFO
Worker: Asked to launch driver

14:43:04 INFO
DriverRunner: Launched

14:43:05 INFO
Utils: Successfully started service 'driverClient'

14:43:06 INFO
ClientEndpoint: Driver successfully submitted

14:43:07 INFO
SparkContext: Submitted application: Spark Pi

14:43:08 INFO
Utils: Successfully started service 'sparkDriver'

14:43:11 INFO
StandaloneAppClient: Connecting to master

14:43:10 INFO
StandaloneSchedulerBackend: Connected to Spark cluster

14:43:12 INFO
Master: Registered app Spark Pi

14:43:13 INFO
Master: Launching executor on worker

14:43:14 INFO
Worker: Asked to launch executor

14:43:15 INFO
ExecutorRunner: Launched

14:43:16 INFO
StandaloneAppClient: Executor added on worker

14:43:17 INFO
StandaloneSchedulerBackend: Granted executor ID

14:43:18 INFO
StandaloneAppClient: Executor is now RUNNING

14:43:19 INFO
SparkContext: Starting job

...

14:43:20 INFO
DAGScheduler: Job finished

14:43:21 INFO
StandaloneSchedulerBackend: Shutting down all executors

14:43:22 INFO
Worker: Asked to kill executor

14:43:23 INFO
ExecutorRunner: Killing process!

14:43:24 INFO
Worker: Driver exited successfully

14:43:25 INFO
Master: Removing app

14:43:26 INFO
SparkContext: Successfully stopped SparkContext

Data Locality in Spark

Sun, 12 May 2019 00:00:00 GMT

This post provides an overview of different types of data locality in Spark. In the coming posts, we'll dive deeper into more low-level concepts. Meaning, we'll explore the Spark internals using examples. Then, we'll explore some examples of running spark applications in a cluster with a standalone cluster manager, YARN cluster manager, and Mesos cluster manager.

Defining Data Locality

In Spark, tasks are run as close to the location of data as possible. Meaning, executors are selected based on their proximity to requested data within the cluster. This notion is referred to as data locality. Since the selection of executors is affected, data locality influences job scheduling as a consequence. To find optimal executors closest to the data, Spark maintains a list of preferred executors for each partition.

Data locality can have a major impact on the performance of Spark jobs. If data and the code operating on it are together, then computation tends to be fast. The goal of data locality is to minimize the read and write speed from the CPU to memory. Data locality is achieved if each HDFS block is loaded in the RAM of the same node where the HDFS block lives. Specifically, data transfer can be avoided if the Spark scheduler runs tasks on executors where these blocks are present.

Levels of Data Locality

The locality level indicates which type of data access is performed. There are five levels of data locality:

PROCESS_LOCAL
- Execute a task on the same executor as the cached data
NODE_LOCAL
- Execute a task on the same node as the cached data, but on different executors
- Generally, this level is slower than the previous one
- However, sometimes there is waiting for available executors
RACK_LOCAL
- Execute a task on the same rack as the cached data, but on different executors and nodes
- Generally, this level is even slower
- This is because even more data is moved through the network
NO_PREF
- There isn't any preference for data locality
ANY
- Execute a task on any executor on any rack, node, or executor

Spark Deployment Modes

Thu, 28 Mar 2019 00:00:00 GMT

This post provides an overview of the different deployment modes in Spark and how each deployment mode changes the behavior of Spark components. In the coming posts, we'll dive deeper into more low-level concepts. Meaning, we'll explore the Spark internals, along with some examples.

An application can be deployed to a cluster in one of two modes: cluster or client mode. These modes determine the location of the driver process. By default, Spark will run a driver in an application on the client JVM. Python applications can't run in cluster mode on a standalone cluster.

To walk through an example demonstrating the interaction between a driver program and cluster components in standalone mode, refer to the my other post.

Client-Deploy Mode

As stated previously, the deployment mode determines the location of the driver process. In client-deploy mode, the driver program runs on the client's JVM process. Meaning, the driver program runs on the client's machine. This is the same machine as the one that called the spark-submit command, which implies the driver process sits outside of the cluster. Generally, applications deployed in client mode perform the following steps:

Client's JVM process submits a driver to the master
The driver is launched
Master instructs workers to start executor processes for the driver
Workers launch executor JVMs
The driver and executors communicate independently
- Doesn't involves the master or workers

Cluster-Deploy Mode

In cluster-deploy mode, the driver program runs on its own JVM process located inside the cluster. Indicating, the driver program doesn't run on the client's machine. Instead, it runs on a node within the cluster. Again, the driver program is started by a worker JVM, but runs in a separate JVM in cluster-deploy mode. Generally, applications deployed in cluster mode perform the following steps:

Client's JVM process submits a driver to the master
Master instructs one of its workers to launch a driver
That worker launches a driver JVM in the cluster
Master instructs any workers to start executors for the driver
Those workers launch executor JVMs
The driver and executors communicate independently
- Doesn't involves the master and workers

Runtime Architecture in Spark

Mon, 11 Mar 2019 00:00:00 GMT

This post provides a high-level introduction to generic objects in the Spark API, along with the responsibilities for each object. In the coming posts, we'll dive deeper into more low-level concepts. Meaning, we'll explore the Spark internals, along with some examples.

Two Common Questions
Driver Program
Spark Context
Spark Scheduler
Cluster Manager
Workers and Executors

The Two Common Questions

A Spark application runs on a cluster with either a standalone cluster manager, YARN cluster manager, or Mesos cluster manager. Although these clusters use very different component comparatively, they still need to answer the same two questions regardless of how each one is set up. Specifically, these questions relate to:

Job scheduling
Resource scheduling

Job scheduling refers to determining which executors run a given set of tasks. To do this, we must know what resources are available, which includes CPU and RAM resources. Job scheduling is typically performed by Spark's scheduler, which will be defined in detail later.

As stated previously, we must know the available resources in order to perform job scheduling. Implying, job scheduling requires resource scheduling, in order to function properly. Resource scheduling refers to determining which executors receive available resources. Specifically, this involves assigning CPU and RAM resources to executors, which represent units of processing in a cluster.

Defining the Driver Program

A driver program represents a Spark program. During the execution of a driver program, it requests for executor processes from the cluster manager. In particular, it requests for CPU and memory resources. Then, it organizes its application components into stages and tasks. The driver program is responsible for defining the tasks sent to executors of a Spark cluster. Then, it collects results from the executors. A driver program is represented as the following objects:

A Spark Scheduler
A Spark Context
A Spark application

Defining a Spark Context

One of the most essential responsibilities of the driver program is to define and initialize a spark context. A spark context acts as the entry point for a driver program to interact with a Spark cluster. In other words, it connects to a spark cluster from an application. A spark context is responsible for configuring a spark session, which is not associated with workers or masters.

Configurations related to a spark session include managing job execution, loading files, and saving files. This also includes assigning the number of executors allocated to an individual application, and includes assigning the number of cores per executor.

Defining the Spark Scheduler

Each driver program comes equipped with a Spark Scheduler, which runs in the background of a driver program. It can be configured via the Spark context. The Scheduler performs job scheduling for its application after communicating with a cluster manager (of a cluster). Indicating, it is responsible for the following:

Sending tasks to executors
Requesting memory and CPU resources from a cluster manager
Deciding which executors run which tasks
Monitoring execution of tasks sent to executors

Notice, the decision of which executors run which tasks depends solely on Spark, rather than the cluster manager. In other words, the Scheduler is responsible for choosing which executors run its tasks, while the cluster manager is only responsible for initializing the executors and keeping them up and running.

Defining the Cluster Manager

Each cluster comes equipped with a cluster manager, which runs on a node in the cluster. In Spark, there are different types of cluster managers that demonstrate their own unique behavior in a cluster. By default, a Spark cluster runs a standalone cluster manager, which is referred to as a Spark master also. Other popular cluster managers include the following:

Standalone: Spark Master
YARN: Application Master
Mesos: Mesos Master

A cluster manager instructs workers to start executor processes when a driver program requests for resouces from the cluster. In order to do this, the cluster manager provides workers with information about the cluster's CPU and memory resources. The cluster manager performs resource scheduling for its cluster. Indicating, it is responsible for the following:

Starting and stops processes in its cluster
Assigning the maximum number of CPUs used by executors
Dividing cluster resources among several applications

In particular, resource scheduling involves distributing the resources of a cluster requested by several applications. Meaning, a cluster manager will make its cluster's resources available to applications via executors. In other words, the cluster manager allocates its cluster's CPU and memory resources to its executors for any applications to use.

Defining Workers and Executors

Each worker runs on a node in its cluster. A worker is responsible for launching executors and monitoring executors for the cluster manager, in case of any failures. Additionally, it is used for launching the driver process when -deploy-mode=cluster. Notice, certain components behave differently depending on the deployment mode of an application. For more information about these behaviors, refer to my next post.

Similarly, each executors runs on a node in its cluster. There can only be one executor per worker. An executor is responsible for accepting tasks from the driver program, executing those tasks, and returning any results to the driver. Each executor has several tasks slots. These slots are used to run tasks in parallel on the executor. Specifically, these tasks are implemented as threads, rather than individual CPU cores. Indicating, they do not correspond to the number of CPU cores on a machine.

Datasets and DataFrames

Sun, 03 Mar 2019 00:00:00 GMT

Describing Spark SQL

Unlike the basic Spark RDD API, the Spark SQL API provides additional data structures used for holding data and performing computations. As a result, Spark SQL is able to perform improved optimizations. Specifically, Spark SQL is used for executing SQL queries. Spark SQL can also be used to read data from an existing Hive installation. When running SQL from within another programming language, the results are returned as a Dataset or DataFrame.

Datasets and DataFrames

Similar to an RDD, a Dataset is a distributed collection of data that can be cached in memory. It provides the following benefits:

Strong typing
Ability to use powerful lambda functions
Optimized execution engine

A DataFrame is a Dataset organized into named columns. It is conceptually equivalent to a table in a relational database or a pandas DataFrame. However, it has improved internal optimizations. For a more detailed description of the benefits of Datasets and DataFrames, please refer to this article.

Visualizing DAGs in Spark

Wed, 27 Feb 2019 00:00:00 GMT

The goal of this post is to provide a general introduction to the RDD API. Each example has a snippet of PySpark code with explanations. Another goal is to provide a general introduction to Spark's web UI. Certain examples have DAG visualizations for jobs and stages. Spark starts a web UI for each SparkContext that is initialized.

This will only include rudimentary examples of methods in the RDD API. For more detailed illustrations and explanations of these concepts, refer to this post and this article. In the coming posts, we'll dive deeper into more generic objects in the Spark API. Then, we'll explore low-level concepts, including the Spark internals.

Setting up a SparkSession
Using an RDD with Python
Counting Words from Files
Visualizing the DAG

Setting up a SparkSession

Before walking through any code examples, we need to create a SparkSession. Each SparkSession acts as an entry point into Spark programming with RDDs. After executing the setup code, we'll be able to use the session in our examples below.

The builder object of a SparkSession provides access to the associated application name, associated master URL, and configuration options. It also provides access to a getOrCreate method, which initializes a SparkSession after setting options in the builder object.

>>> from pyspark.sql import SparkSession
>>> spark = SparkSession \
...         .builder \
...         .appName('TxtRdr') \
...         .getOrCreate()
>>> sc = spark.sparkContext

Some examples use a set of files named hello.txt and order.txt. A SparkContext will read these files using the textFile method. After creating an RDD, we'll call methods to transform and filter the data in the files listed below.

hello.txt:

Hello world
This is a file

order.txt:

Hello friend
This file is my order
Burger, fries, soda

Notice, there are two similar words found in both files. Specifically, these words are Hello and file. In the upcoming examples, we'll run some standard Python code and Spark code to find these words. Then, we'll count the number of each word using Spark.

Using an `RDD` with Python

Before performing any transformations, the data needs to be read using the textFile method. Specifically, the SparkContext will read the hello.txt and orders.txt files using this method, which returns an RDD object. After calling this method, the returned RDD essentially contains a list of strings, where each string represents a line from the file.

The flatMap method maps a function that is defined using the lambda keyword to our RDD. In this example, our function separates each line into individual strings based on any spaces. Although it isn't used here, the map method is very similar to the flatMap method. Specifically, the map method performs an extra step, which involves storing these individual strings into a list for each line.

Ultimately, the collect method is called, which returns all the elements from the RDD as a list to the driver program. This method is an action. In other words, the transformations prior to the collect method aren't computed until the collect method is called. The code snippet below creates an RDD and performs the described operations on the hello.txt file.

>>> hello = sc.textFile('hello.txt') \
...           .flatMap(lambda c: c.split(' '))
...           .collect()

For this example, there's still a second file with words that should be counted by the same Spark program. Through a similar process as before, the collect method is called for reading in the order.txt file. Additionally, the other methods are called for function mapping and data collection.

>>> order = sc.textFile('order.txt') \
...           .flatMap(lambda c: c.split(' ')) \
...           .collect()

After executing these two code snippets, the resulting RDD includes a long list of each word in both files. Since the collect method outputs a list from the RDD, we can perform standard Python functions on it. As an example, we can use Python to find the common words between the two RDDs.

>>> list(set(hello) & set(order))
['Hello', 'file']

Counting Words from Files

Now, we may prefer to compute some of these operations in Spark. By performing filters in CPython, we will lose the benefit of distributed computation in Spark. If hello.txt and orders.txt are much larger files, or if we instead count the words of thousands of files, filtering in Spark becomes much faster compared to filtering in ordinary CPython on a single machine.

To illustrate this point, let's go through an example that is very similar to the previous one, but ultimately performs counting and filtering.

>>> hello = sc.textFile('hello.txt')
>>> order = sc.textFile('order.txt')
>>> counts = hello.union(order) \
...               .flatMap(lambda w: w.split(' ')) \
...               .map(lambda c: (c, 1)) \
...               .reduceByKey(add) \
...               .filter(lambda t: t[1] > 1) \
...               .collect()

Again, calling the flatMap method returns an RDD containing individual words as strings. The map method converts each string into a tuple containing words and their frequency. Then, the reduceByKey operation loops through each key-value pair and adds up the values for any repeated keys. Lastly, the filter method will only include key-value pairs, where the value is greater than one.

>>> counts
[('Hello', 2), ('file', 2)]

As expected, the program outputs two key-value pairs. The first key-value pair is ('Hello', 2), since there are two of these keys in both files. Similarly, the second key-value pair is ('file', 2), since there are two of these keys in both files.

Visualizing the DAG

The number of jobs for an application equals the number of RDD actions. In our example, we can see there is a single job. This is because there is only one action, which is collect. Instead of having a single job, we would have two jobs with this code:

>>> hello = sc.textFile('hello.txt').collect()
>>> order = sc.textFile('order.txt').collect()

The number of additional stages equals the number of wide transformations in an application. In our example, we can see there are two stages in total, but only a single additional stage. This is because there is only one wide transformation, which is reduceByKey. Notice, the web UI in Spark provides a nice visualization of the DAG.

Recall, the number of tasks within a stage equals the number of partitions in an RDD. By default, Spark assigns the number of paritions to be two. Meaning, the default number of tasks per stage is two. The number of partitions is assigned to an RDD when it is initialized. Thus, this parameter can be adjusted when calling textFile(file, num_partitions). Therefore, our stage will have six tasks (instead of four) if we change this line of code:

>>> hello = sc.textFile('hello.txt', 6)

Since Spark assigns the number of paritions of an RDD during initialization, the number of tasks are determined after shuffling as well. This is because Spark creates a new RDD after shuffling. This object is called a ShuffleRDD. For a more detailed description of how tasks are separated and organized in Spark, refer to this post.

Spark RDD Fundamentals

Fri, 15 Feb 2019 00:00:00 GMT

This post provides a high-level introduction to the RDD object in the Spark API. In the coming posts, we'll dive deeper into more generic objects in the Spark API. Then, we'll explore low-level concepts, including the Spark internals.

Resilient Distributed Datasets
Defining a DAGScheduler
Types of Transformations
Lifecycle of a Spark Program

Resilient Distributed Datasets

Most likely, we've all worked with pandas DataFrames before. They're in-memory, single-server data structures that offer many user-friendly functions for data processing. Functionally, Spark provides a data structure that is very similar in this sense, but can be used across multiple servers. This data structure is called a resilient distributed dataset, or an RDD for short. In short, an RDD is an in-memory data structure that is distributed across many servers within a Spark cluster.

Roughly, we can think of an RDD as a distributed version of a pandas DataFrame. I'm making this comparison because RDDs offer many pandas-like functions that are focused around data processing. These functions are called Transformations and Actions. Specifically, transformations create a new dataset from an existing one. Contrastingly, actions return non-dataset values, which generally relate to some aggregation.

Transformations are lazy. Meaning, an RDD isn't computed until it receives an action. Actions always return values to the driver program. Spark receives a performance boost from any lazy evaluations. However, this could become a problem if users continuously recompute that same transformation. As a result, Spark allows us to persist an RDD to memory using the persist method. To summarize, transformations and actions have the following properties:

Transformations are lazy by default
Actions aren't lazy
Transformations return a new RDD
Actions return an aggregated value of the RDD

As stated previously, an RDD receives a huge boost in performance by keeping data in memory. However, Spark supports data persistance to disk as well. Spark also supports data persistence to databases. To summarize, an RDD has the following properties:

Distributed
Fault-tolerant
Flexible functions such as map(func)
Optionally in-memory on the Driver's JVM
Parallelizable using sc.parallelize(data)
Immutable (more on this later)

Defining a `DAGScheduler`

Recall, a transformation is a type of special RDD method that returns another RDD object. Again, these methods aren't computed until the RDD receives an action, indicating that RDD objects are immutable. Since RDD objects can't change once they are created, Spark creates a new object called a DAG when an action is called. In a DAG, each node is an RDD partition, and an edge is a transformation.

Spark breaks the RDD into smaller chunks of data called partitions. In Spark, a partition is a chunk of data on a node within the cluster. At a high level, Spark breaks transformations and actions into Tasks, which are mapped to partitions. Essentially, a Task represents a unit of work on a partition of a distributed dataset.

Assuming nonsequential dependence, Tasks are executed in parallel on partitions. Thus, the number of partitions made up by an RDD should equal the number of CPU cores within a cluster. Theoretically, increasing the number of partitions would increase the amount of parallelism for a system, assuming there are available CPU cores. If an RDD can't fit an entire set of data into memory, then that data will be stored to and read from disk.

Now, let's return to our previous discussion about the DAG object. When an action is called, each DAG is submitted to a DAGScheduler object for execution. A DAGScheduler organizes operations into Stages, and a Stage is organized into Tasks. Each Task is scheduled separately. It represents a unit of work on a partition of an RDD, and is executed as a thread in an executor's JVM. The DAGScheduler returns a TaskSet object, which is passed to a TaskScheduler. The TaskScheduler launches tasks in the a cluster manager.

Multiple tasks can be executed in parallel for any stage. Specifically, any two stages can be executed in parallel if they aren't sequentially dependent on each other. Implying, tasks from one stage can be executed in parallel with tasks from a separate stage, if they aren't sequentially dependent on each other. Refer to this post for an illustration of how Tasks and stages run in parallel.

The number of tasks is equal to the number of partitions. The number of stages is equal to the number of wide transformations. For examples that may help illustrate these concepts, refer to my next post.

Types of Transformations

There are two types of transformations that can be applied to RDDs: narrow transformations and wide transformations. Narrow transformations refer to transformations where each partition contributes to one stage only. These include transformations like map, filter, etc. Wide transformations refer to transformations where each partition contributes to many stages. In Spark, this concept is called shuffling.

Shuffling is used for regrouping data between partitions. Shuffling is necessary for situations requiring information from each partition. Wider transformations are more expensive than narrow transformations in comparison. For example, the map transformation doesn't require shuffling, since it applies element-wise transformations to each partition. This technique is called pipelining. In other words, an element in one partition doesn't need any information from other partitions. On the other hand, the groupByKey wide transformation needs information from each partition. Specifically, a narrow transformation keeps its results in memory, whereas a wide transformation writes its results to disk. This post defines optimized shuffling algorithms in detail.

Lifecycle of a Spark Program

Now, we have a high-level level understanding of the core Spark data structures. This lecture defines the lifecycle of a Spark program in detail. Generally, a common lifecycle of a spark program looks like the following:

Create some input RDDs from external data
Lazily transform them to define new RDDs using transformations
Ask Spark to cache() any intermediate RDDs that will be reused
Launch actions to start parallel computation
Spark optimizes and executes its computations

Hadoop as a Distributed OS

Mon, 22 Oct 2018 00:00:00 GMT

Before investigating Spark in detail, we should develop a core intuition behind Hadoop. This post compares Hadoop to a traditional computer operating system. In the coming posts, we'll begin exploring generic objects in the Spark API. Then, we'll dive deeper into more low-level concepts, including the Spark internals.

Developing a core intuition behind Hadoop is an important first step before investigating Spark in great detail. Recall, a basic computer operating system constists of two essential components:

A file system
A scheduler

As a reminder, a file system manages user data. A scheduler manages any running process or program in the system. These programs involve storing, retrieving, or updating the data in the file system.

Roughly, Hadoop can be seen as a distributed operating system. In this comparison, YARN represents a distributed scheduler. Similar to a scheduler in a basic operating system, YARN does the following:

Monitors computing resources
Schedules jobs involving processing

However, the key difference is YARN performs these functions across many different machines. In a similar fashion, HDFS differs from a standard file system, because it manages user data across many different machines. Lastly, MapReduce programs are the distributed form of programs in a traditional computer system.

Relevance of Hadoop

Thu, 17 May 2018 00:00:00 GMT

It's been 14 years since the initial release of Apache Hadoop, which is a long time for any software. Unsuprisingly, the internet is flooded with clickbait articles about Hadoop being replaced by newer and improved alternatives. This isn't to say these opinions don't have any merit. Even now, Hadoop remains essential for many of today's largest enterprises with their own data centers. However, other companies have the liberty to outsource sensitive data to the cloud. In this case, cloud storage services, such as AWS S3, can easily replace HDFS in the cloud.

Recall, Hadoop has two other frameworks in its ecosystem: the YARN resource manager and the MapReduce computing paradigm. Recently, some companies have replaced YARN with Kubernetes for scheduling, and most have replaced MapReduce with Apache Spark for computing. Specifically, Apache Spark excels when used in the cloud and data centers, contributing to its wild popularity. Hopefully, this serves as motivation to learning more about Spark.

This post hopefully motivates the use-cases and benefits of Spark in today's world. The comping posts will provide a high-level introduction to generic objects in the Spark API. Then, we'll dive deeper into more low-level concepts, which includes the Spark internals.

Running Hugo on GitHub

Tue, 10 Apr 2018 00:00:00 GMT

In this post, we walk through the steps of running a site on GitHub created by a static site generator. This post assumes a directory containing hugo source files has already been created. For more information about hosting a project on GitHub as a submodule, refer to this article.

Access the local hugo directory
- We'll refer to this directory as ./mysite
- This directory will contain the files on the hugo site
- The directory should look like this:

mysite/
├── archetypes/
├── config.toml
├── content/
├── data/
├── layouts/
├── public/
├── resources/
├── static/
└── themes/

Create a new repository in GitHub
- Call this repository <username>.github.io
- Where <username> is a github username
- This repository will contain the rendered version of the site
Call hugo to create a public directory
- This command build the site to the public directory
In the new public directory, initialize the git repo

$ cd public
$ git init
$ git add .
$ git remote add origin https://github.com/<username>/<username>.github.io.git
$ git commit -m "Commit to site"
$ git push origin master

Darius Kharazi's Blog

An Open-Source ML Pipeline

Using Feast as a Feature Store

Using Great Expectations for CI/CD Data Validations

Using MLflow for Model Registry and Experiment Tracking

Using Hyperopt for Distributed Hyper-Parameter Tuning

Using Evidently for Model Monitoring

Goodhart's Law

The Dead Sea Effect

Comparing Common Implicit Recommenders

Comparing the ALS and BPR Losses

Comparing the BPR and WARP Losses

Improving Health Care with 1% Steps

The Problem with Out-of-Network Billing

The Problem with Current Provider Prices

The Problem with Current Adjudication for Health Insurance Claims

References

Outlining the A/B Testing Procedure

Defining a Problem Statement

Designing a Hypothesis Test

Designing an Experiment

Running the Experiment

Validating the Experiment

Interpreting the Results

Launching a Decision

Intuition behind BTYD Models

Potential Reasons for Recent Shortages

Six Dimensions of National Culture

Illustrating the Axis for Collectivism

Illustrating the Axis for Power Distance

Illustrating the Axis for Masculinity

Illustrating the Axis for Uncertainty Avoidance

Illustrating the Axis for Short-Term Orientation

Illustrating the Axis for Indulgance

Collider Bias and Police Use of Force

The Search for Answers

A Glimpse into Fryer's Findings

Potential Issues with Collider Conditioning

Gram Matrices in Neural Style Transfer

An Argument against Meat

Economics of Livestock Production

Health Effects from Livestock Farming

Climate and Deforestation Impact from Animal Agriculture

Ethics of Livestock Farming

Understanding the Glicko Rating System

Comparing Elo with Glicko

Intuition behind Uncertainty of Ranking

Intuition behind Expected Ranking

Building a Prototyping Pipeline in Python

Table of Contents

Motivating a Data Science Pipeline

Outlining a Prototyping Pipeline

Illustrating our Project Layout

Defining a Flask Application

Defining NGINX Configurations

Defining Gunicorn Configurations

Defining the Dockerfile

Selection Bias and COVID-19

State of COVID-19 Testing

Converse of Survivorship Bias

Creating Custom Awaitable Objects

Motivating the Await Expression

Introducing Awaitable Objects

Defining an Awaitable Object

Internal Structure of Pandas DataFrames

Table of Contents

What is a BlockManager

Illustrating the BlockManager

Illustrating the Role of the BlockManager

Benefit of the BlockManager

Disadvantages of the BlockManager

Roadmap for the BlockManager

Performance Benchmarks: PyArrow

Table of Contents

The Origins of PyArrow

Benefits of the Arrow Protocol

Use Cases of PyArrow

Comparing PyArrow to Parquet

Why Serialization can be a Problem

Improvements to toPandas

Improvements to `toPandas`