squash commits in git

Mingmin

When submitting a pull request, it’s highly suggested to squash your commits down to a few, or one, discreet changesets.

To do that, the below example shows the detailed steps:

Now I have added three commits in a Git repository as below
three original commits

Next, let’s start the squash. First go to the home folder of git repository. Then run the command

# there're 3 commits, so `HEAD~3` here
git rebase -i HEAD~3

You can see below prompted message

pick 37ff3bd commit1
pick 32756c4 commit2
pick a13ca74 commit3

Change the word pick to squash except the first line, like

pick 37ff3bd commit1
squash 32756c4 commit2
squash a13ca74 commit3

After save a new commit is ready, and you can update with another commit comment like:

squash 3 into 1
# This is a combination of 3 commits.
# This is the 1st commit message:
commit1

# This is the commit message #2:

commit2

# This is the commit message #3:

commit3

Now run the git push command

git push --force

Now in git repository, you would find that the previous 3 commits are replace by one squashed commit.
squashed commit

The back-end of FlinkRunner in Apache Beam

Mingmin

Table of Content

1. Beam SDK-Runner model

The Beam SDKs provide a unified programming model that can define your data processing pipeline, regardless how it’s actually processed. The Beam Runners translate the data processing pipeline into the API compatible with the distributed processing back-end of your choice.

To make is simple, Beam SDK defines a context-free data processing model. When it goes to an actual execution engine, like Flink, Spark, Storm, event MapReduce, the Runner is responsible to do the translate work, to re-express the data pipeline with its own API.

2. PipelineRunner interface

PipelineRunner is the parent class for any runner. It contains a core function run as below, which is responsible to translate a Pipeline in Beam SDK to runner API.

  /**
   * Processes the given Pipeline, returning the results.
   */
  public abstract ResultT run(Pipeline pipeline);

3. FlinkRunner

Apache Flink is an open-source stream processing framework for distributed, high-performing, always-available, and accurate data streaming applications.

FlinkRunner is the one to run a Beam Pipeline on a Flink cluster.

3.1. FlinkPipelineExecutionEnvironment

In Beam SDK, there’s no difference for both bounded and unbounded data API, which is not the same in Flink. FlinkPipelineExecutionEnvironment hides the difference, inside of it, there’s org.apache.flink.api.java.ExecutionEnvironment for bounded data, aka batch process; and org.apache.flink.streaming.api.environment.StreamExecutionEnvironment for unbounded data, aka streaming process.

3.2. Translate Pipeline to Flink DataStream

Now let’s go to the core function public PipelineResult run(Pipeline pipeline) {. The main logic is expressed with below lines:

FlinkPipelineExecutionEnvironment env = new FlinkPipelineExecutionEnvironment(options);

env.translate(this, pipeline);

JobExecutionResult result = env.executePipeline();

...

public void translate(FlinkRunner flinkRunner, Pipeline pipeline)

The first step is to detect the mode of pipeline, whether it’s STREAMING or BATCH. It’s quite straight-forward with below lines

TranslationMode translationMode = optimizer.getTranslationMode();
...
if (options.isStreaming()) {
  return TranslationMode.STREAMING;
}

If it’s TranslationMode.STREAMING, a FlinkStreamingPipelineTranslator is created, otherwise FlinkBatchPipelineTranslator for TranslationMode.BATCH mode.

    if (translationMode == TranslationMode.STREAMING) {
      this.flinkStreamEnv = createStreamExecutionEnvironment();
      translator = new FlinkStreamingPipelineTranslator(flinkRunner, flinkStreamEnv, options);
    } else {
      this.flinkBatchEnv = createBatchExecutionEnvironment();
      translator = new FlinkBatchPipelineTranslator(flinkBatchEnv, options);
    }

The details to setup environment from FlinkPipelineOptions will be described later, let’s focus on the pipeline translator first.

Take TranslationMode.STREAMING mode for example. FlinkStreamingPipelineTranslator extends Pipeline.PipelineVisitor.Defaults, and leverage the traverseTopologically to perform the detailed transform work. The PipelineVisitor is very important to understand the implementation of Runner, will update another post to talk it later soon.

submit to run

Now a pipeline job expressed with Flink DataStream is ready, it can be submitted to run with method flinkEnv.execute(jobName);

3.3. usage of FlinkPipelineOptions

All pipeline options can be found in JDK API FlinkPipelineOptions. I’ll highlight several ones that are related with the implementation of runner itself.

–streaming

As mentioned above, FlinkRunner has separated code for batch and streaming, so make sure you use the right one.

-d in bin/flink run

Mostly I don’t submit a Flink job as standard Java application, instead bin/flink run is used. By default it waits until job is finished. For batch, it’s OK, while in streaming mode, it’s highly suggested to add a -d option in the CLI command. With it, it’s running in Detached mode, and return immediately.