I think you get my point now. API errors occur quite often. Most of them are due to temporary service unavailability, caused mainly by having too much traffic at the moment. The simple solution is to retry. In this post, I’ll show how to easily implement efficient retry mechanism.

According to Wikipedia

Exponential backoff is an algorithm that uses feedback to multiplicatively decrease the rate of some process, in order to gradually find an acceptable rate.

In case, which I’ll explore, it’s a way to retry a piece of code, in case of an exception occurrence. It will delay every attempt with exponential pause, according to the equation (2^attempt) * baseTime

Below example uses PHP’s backoff library. Similar libs can be found in Python, node.js and probably any language of choice.

use GuzzleHttp\Exception\RequestException;
use Monolog\Logger;
use Monolog\Handler\StreamHandler;
use Monolog\Processor\UidProcessor;
use STS\Backoff\Backoff; # nothing to do with sts.pl ;)

$log = new Logger('x');
$log->pushHandler(new StreamHandler('php://stdout', Logger::DEBUG));
$log->pushProcessor(new UidProcessor());

$client = new GuzzleHttp\Client();
// below API fails 60% of time
// always fail first call in 5 minutes timespan
$url = 'https://sznapka.pl/fakeapi.php';
$log->debug(sprintf('Fetching from %s', $url));

$backoff = new Backoff(10, 'exponential', 10000, true);
$result = $backoff->run(function() use ($client, $url, $log) {
    try {
        $res = $client->request('GET', $url);
        $data = json_decode($res->getBody(), true);
        $log->info(sprintf('Got response, %d items', count($data)));
    } catch (RequestException $e) {
        $log->error($e->getResponse()->getBody());
        throw $e; // causes backoff lib to retry
    }
});
$log->debug('All done');

When you run it with php ./console.php you’ll get:

[22:29:15.649] x.DEBUG: Fetching from https://sznapka.pl/fakeapi.php [] {"uid":"6b3d944"}
[22:29:15.759] x.ERROR: API failed, what a surprise! [] {"uid":"6b3d944"}
[22:29:15.832] x.ERROR: API failed, what a surprise! [] {"uid":"6b3d944"}
[22:29:16.268] x.INFO: Got response, 5 rows [] {"uid":"6b3d944"}
[22:29:16.269] x.DEBUG: All done [] {"uid":"6b3d944"}

As you can see, the backoff library retries with exponential intervals until our wrapped closure doesn’t throw an exception. It has set 10 retries and waitCap 10 seconds, so will stop processing whenever one of those conditions appear. I’ve also defined jitter parameter to true to spread out retries and minimize collisions.

Last but not least – always log your external API usage. It will help you a lot during the debugging phase. Also, it’s very handy to use UidProcessor which puts an uid (per session) into logger’s context. It allows filtering logs from given invocation, which is especially helpful with overlapping calls or concurrent usage.

One of the least used and least appreciated features, in my opinion, is repeated fields. The name doesn’t indicate well enough the intention, so for a sake of simplicity please consider it as an array field or nested field. You can define any structure inside the repeated field you like, leveraging types of columns which regular columns can be. The important part is to set mode REPEATED for the field of type RECORD.

In below example, you’ll see the table definition and SQL INSERT query, which will populate it.

insert into test.oscars values  
  ('Best Actor', 2019, [
    struct('Joaquin Phoenix', 'Arthur Fleck / Joker', 'Joker', true),
    struct('Antonio Banderas', 'Salvador Mallo', 'Pain and Glory', false),
  /* .. */
  ]),
  ('Best Actor', 2018, [
    struct('Rami Malek', 'Freddie Mercury', 'Bohemian Rhapsody', true),
    struct('Christian Bale', 'Dick Cheney', 'Vice', false),
    /* .. */
  ]));

Now, you can query this table and immediately you’ll see a benefit of it – the records will be flattened. What’s worth noting – if you use BigQuery SDK for any language or even bq CLI tool, you’ll get 3 records and every one of them will have our nominee field.

Now, bear in mind, that our example is very simple. Imagine that in real-world scenarios, you’ll have 100+ columns in the repeated field, but you only need a few. Also, due to column-oriented nature of underneath storage in BigQuery, it’s always the best to limit the number of columns (see: Control projection – Avoid SELECT *).

Let’s say, we want to get only an award and year and the actor who has been nominated and the flag for the winner. My first bet was to use:

SELECT award, year nominees.actor, nominees.winner
FROM test.oscars

Wrong! It turned out, it’s not the way how it works. You need to use ARRAY mechanism and UNNEST the nominees in order to subquery it and return a RECORD in the form you need. Here goes:

SELECT
  award,
  year,
  ARRAY(
    SELECT STRUCT(actor, winner)
    FROM UNNEST(nominees)
    ORDER BY winner DESC, actor
  ) AS nominees
FROM test.oscars
ORDER BY year

By using ARRAY construct you can leverage full query on repeated field which gives you a lot of flexibility. Hope you’ll find it out useful!

Cover photo created by natanaelginting – www.freepik.com

But what happens when you start it on production and need to have some historical data, in order to keep consistency? Of course – there is a need to import that. There are many ways to achieve that, including writing some custom script. I urge you to have a look at redis --pipe option, also called Redis Mass Insertion, where you can leverage Redis’ protocol in order to really quickly ingest a lot of data (way faster than writing a custom script to migrate data using Redis SDK).

Once you have redis-cli tool, prepare a file with Redis commands and run redis-cli --pipe:

$ cat migration.txt
SET migrated-test-1 value-1
SET migrated-test-2 value-2
SET migrated-test-3 value-3

$ cat migration.txt | redis-cli  --pipe
All data transferred. Waiting for the last reply...
Last reply received from server.
errors: 0, replies: 3

$ redis-cli get migrated-test-1
"value-1"

In case you want to use different data structure provided by Redis, you can also do that. Let’s take an example Sorted Sets:

$ cat migration-sortedsets.txt
ZADD migrated-ss-1 1 value1
ZADD migrated-ss-1 2 value2
ZADD migrated-ss-2 1 value3
ZADD migrated-ss-2 2 value4

$ cat migration-sortedsets.txt | redis-cli  --pipe
All data transferred. Waiting for the last reply...
Last reply received from server.
errors: 0, replies: 4

$ redis-cli zrange migrated-ss-1 0 -1
1) "value1"
2) "value2"

$ redis-cli zrange migrated-ss-2 0 -1
1) "value3"
2) "value4"

Everything seems perfect so far. But how to quickly create those import files? I often use a dirty trick, which is not the most elegant but works like a charm. Use CONCAT, which is being provided by every SQL engine:

$ mysql> select * from games;
+----------------+----------------+------------+
| studio         | game           | popularity |
+----------------+----------------+------------+
| CD Project Red | Witcher III    |          1 |
| CD Project Red | Cyberpunk 2077 |          2 |
| EA Sports      | Fifa 21        |          1 |
| EA Sports      | NHL 21         |          2 |
+----------------+----------------+------------+
4 rows in set (0.00 sec)

Create query which will transform your data into Redis commands:

SELECT
  CONCAT('ZADD games-', replace(studio, ' ', '_'), ' ', popularity, ' "', game, '"')
FROM
  games;

Now run the query and save result to file and ingest it:

$ cat query.sql | mysql -uuser -ppass dbname 2>/dev/null | tail -n +2 > redis-import.txt

$ cat redis-import.txt
ZADD games-CD_Project_Red 1 "Witcher III"
ZADD games-CD_Project_Red 2 "Cyberpunk 2077"
ZADD games-EA_Sports 1 "Fifa 21"
ZADD games-EA_Sports 2 "NHL 21"

$ cat redis-import.txt | redis-cli --pipe
All data transferred. Waiting for the last reply...
Last reply received from server.
errors: 0, replies: 4

Our data is here!

$ redis-cli zrange games-CD_Project_Red 0 -1
1) "Witcher III"
2) "Cyberpunk 2077"
$ redis-cli zrange games-EA_Sports 0 -1
1) "Fifa 21"
2) "NHL 21"

Please note, that for larger data sets Redis might refuse to ingest it, so I suggest to split into smaller files and import them in bash for-loop.

Full story: around 6 years ago I created a simple tool to speed up my daily job. The problem was – our Affiliate Manager has been giving me excel file with one column – the IDs of customers to change their affiliate association in the database. There was a simple query behind it:

UPDATE clients
SET affiliate_id = 100001
WHERE id IN (<here goes comma separated list of clients>);

Of course, you can do that somehow in Excel. I often pasted that column to VIM and put the commas using a macro. But that was becoming a hassle, when I’ve been asked a couple times per week, sometimes a day.

I decided to create a simple tool, which looked ugly, but worked just fine.

I’ve changed job since then, but still use commarize.com almost every day. I’ve seen people in my teams having the same problem so I’ve given them the tool as well. Today is a good time to open-source it and make it slightly prettier, thanks to Tailwind CSS.

I’ve also pushed source to the GitHub, feel free to use it your own way or submit Pull Request or Issue: https://github.com/wowo/commarize

• it’s one of most size efficient (compared to JSON, protobuf, or parquet); AVRO serialized payload can be 10 times smaller than the JSON equivalent,
• enforces usage of a schema,
• works out of the box with Kafka Connect (it’s a requirement if you’d like to use BigQuery sink connector).

Let’s see how to send data to Kafka in AVRO format from PHP producer, so that Kafka Connect can parse it and put data to sink.

I assume you know the basics of AVRO and Schema Registry, but if not – let me know in the comments, I’d be happy to help with setting up Schema Registry! Footnote: the easiest way is it, is to use Confluent’s official Docker image and deploy it on Kubernetes.

In order to use PHP producer which will serialize payloads in AVRO, we need to send it in a particular „envelope”, which contains schema ID. This way Kafka Connect will know which schema should be retrieved from Schema Registry.

Let’s consider, that our PHP Kafka producer will send simple payload:

$data = [
  'time' => '2020-09-24 20:45:00',
  'level' => 'info',
  'channel' => 'main',
  'message' => 'Some log entry has been produced',
];

First of all, you need to create an AVRO schema for that. You can get one by using number of tools available online, like this.

Our AVRO schema will look like this:

$schema = <<<SCHEMA
{
  "name": "LogEntry",
  "type": "record",
  "namespace": "pl.sznapka",
  "fields": [
    {
      "name": "time",
      "type": "string"
    },
    {
      "name": "level",
      "type": "string"
    },
    {
      "name": "channel",
      "type": "string"
    },
    {
      "name": "message",
      "type": "string"
    }
  ]
}
SCHEMA;

Next step is to register our AVRO schema in Schema Registry and obtain its ID. You can either call Schema Registry API directly or use PHP library flix-tech/confluent-schema-registry-api (which I recommend).

Note – you should register your schema under Kafka topic name with suffix ‚-value’

$kafkaTopicName = 'logs';
$subject = $kafkaTopicName . '-value';
$schemaRegistry->register($subject, \AvroSchema::parse($schema));

Once your schema is in the Schema Registry you need to retrieve ID for your subject:

$schema = $schemaRegistry->latestVersion($subject);
$id = $schemaRegistry->schemaId($subject, $schema);

The last part is to create AVRO serialized payload with header, which will be readable for Kafka Connect:

$io = new \AvroStringIO();
$io->write(pack('C', 0)); // magic byte - subject version
$io->write(pack('N', $id));
$encoder = new \AvroIOBinaryEncoder($io);
$writer = new \AvroIODatumWriter($schema);
$writer->write($data, $encoder);

$kafkaPayload = $io->string();

Now we have $kafkaPayload which we can send to Kafka (using https://github.com/arnaud-lb/php-rdkafka) and then it can be distributed to any sink, which you’ll register in Kafka Connect. In my case it’s often BigQuery and Redis.

$topic = $kafkaProducer->newTopic($kafkaTopicName);
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $kafkaPayload);
$kafkaProducer->flush(1000);

That’s it, happy streaming!

For such volumes, traditional tools like Relational Database Management Systems are no longer suitable. Terabytes or even petabytes are quite common numbers in big data context, which is definitely not the capacity that MySQL, PostgreSQL, or any other database can pick up.

To harness huge amounts of data, Apache Hadoop would generally be the first and natural choice, and it’s probably right, with one assumption: Apache Hadoop is a great tool for batch processing. It proved to be extremely successful for many companies, such as Spotify. Their recommendations, radio, playlist workloads, etc. are suitable for batch processing. However, it has one downside – you need to wait for your turn. It usually takes about one day to process everything, scheduled accordingly and executed in a fail-over manner.

But what if we don’t want or can’t wait?

In this instance, streaming technology comes to the rescue. Everything started with Apache Storm project, released in September 2011 and later acquired by Twitter. It’s still a significant player in the streaming market, but nowadays Apache Spark and its streaming module have gained incredible popularity. Spark Streaming provides scalable, high-throughput, and fault-tolerant stream processing of live data streams.

Unlike Apache Storm, which is purely a stream events processor, Spark can be combined with streaming of other libraries, such as machine learning, SQL or graph processing.

That gives endless possibilities and use case coverage.

Overall, Spark Streaming doesn’t differ that much from regular Spark batch jobs. In both cases, we operate on some input, apply transformations and/or compute something out of our input data and then output it somewhere. The only difference is the continuous character of streaming jobs – they run indefinitely until we terminate them (just like a stream of water in a river compared to a bucket filled with water, as an analogy to batch processing). That said, we can choose one of the sources of our streams:

• Apache Kafka
• Apache Flume
• HDFS or S3 filesystem
• Amazon Kinesis
• Twitter
• TCP socket
• MQTT
• ZeroMQ

Custom source (not available for Python yet, just Java and Scala).
As seen in the list above, there’s quite a rich choice of inputs. From those above, two are really fascinating: Kafka and Flume.

Apache Kafka is a natural way forward when one needs to deal with a huge throughput. Designed by LinkedIn engineering team, it’s capable of handling millions of requests per second and can guarantee the order of messages in the queue. It was built with scalability in mind and it is one of the easiest ways to integrate with Spark Streaming. What’s worth highlighting – Kafka comes very handy in terms of installation and publishing/subscribing (provides binding in the most popular languages).

On the other hand, it often happens that sources of streams are very various and have various formats. In this case, Apache Flume is an intelligent choice. It is a standalone Linux program, which takes care on collecting and moving large amounts of data. The concept is quite similar to Spark Streaming, with one exception – it’s only responsibility is to move data from a source to a sink.

Flume supports various sources, like execute command output, directory spool, Kafka, Syslog, HTTP. There’s also a decent choice of sinks, like HDFS, file roll, Kafka, ElasticSearch to name just a few. But the power of Flume rests in custom sinks and sources. One of them is custom Spark Sink, which is used to either push data to Spark directly (through Avro buffer) or expose that data, so Spark can pull it. There’s an impressive list of Flume plugins, be it sinks or sources, which allows you to connect multiple sources to Spark sink for further stream processing.

There are few points to keep in mind working with streaming as opposed to regular batch processing:

• Stateful operations – each DStream (discretized stream) consists of RDDs carried in defined intervals. We can process them with pretty much the same API functions as normal Map-Reduce tasks. The only difference is that we start from a blank point in every interval. To relate to data processed in previous intervals, the state functions need to be applied on the stream (mapWithState or updateStateByKey),
• Window operations – they can be applied over a sliding window of data to apply some transformations,
• Check-pointing – in order to ensure resilience and fault-tolerance, Spark Streaming comes with check-pointing, useful for recovering from node failures,
• Machine Learning operations – MLlib can be used in streaming jobs and out there are especially streaming algorithms (like Streaming Linear Regression or Streaming KMeans). For other applications, traditional MLlib models, already created based on historical values could be used to be applied to streaming data.

As seen above, Apache Spark is a tremendously powerful way to deal with streaming workloads. It leverages common API for RDD transformations which eases the way for achieving clean and coherent Lambda Architecture and allows to utilize Apache Spark set up for all kinds of big data processing.

Now, if we all agree that Type Hinting is the way to go, what should we put in method signatures? We have few options: concrete class, base class or an interface. It all depends on situation. The most flexible way is the interface, because it’s small definition of object behavior and one class can implement multiple interfaces. Moreover, interfaces can be very easily mocked (by both mock tools, like Mockery or by mocks written by hand). All those adds up into great flexibility

Other option is to set class as type hint. You’re limited to this class instances and their descendants. Having multi-tier hierarchy graph, it’s often good idea to use a class in hierarchy near the root or even abstract class, which gives us possibility to apply method to whole graph of inheritance.

The least flexible way is to set a concrete class (near to final or not ever extended in current sytem). In such case you limit method only to serve for such objects, which is understandable, when method does a specific job.

Three options were presented above (interface, base class and concrete class). There’s one more, very important thing, that need to be kept in mind. Although PHP allows you to call methods outside the type that is defined for the method, you should never do that! It can lead to some bizarre errors and breaks Liskov Substitution Principle. Simply said: if you use a type in method’s signature, then it should rely only on this type, not it’s descendants (even we know about their existence), so you can substitute give type with new subclass without altering method body.

Have a look at possible violation of Liskov Substitution Principle:

class UserRepository extends \Doctrine\ORM\EntityRepository
{
    public function findActiveUsers()
    {
        // do some query to retrieve the result
        return $activeUserCollection;
    }
}

// .. 

public function notifyActiveUsers(EntityRepository $repo)
{
    if ($repo instanceof UserRepository) {
         $usersCollection = $repo->findActiveUsers();
    } elseif ($repo instanceof ManagersRepository) {
         // .. do someting else
     }
    // do something with $usersCollection
}

As we can see, type hinted method notifyActiveUsers internally relies on specific extension of EntityRepository. This breaks LSP and leads to unreadable model. Even worse situation can be following:

public function notifyActiveUsers(EntityRepository $repo)
{
    $usersCollection = $repo->findActiveUsers(); 
    // do something with $usersCollection
}

In the stage of implementation, we knew, that only EntityRepository was passed to this method. But in some random point of time, someone can pass other implementation of EntityRepository, and we have a problem. In Java, compiler wouldn’t allow you to compile such code, but in PHP it’s allowed on interpretation stage, however it will fail during runtime.

To sum things up: Type Hinting is an essential concept in OOP and should be used whenever you pass an object as a method parameter. The most flexible way is to use interfaces, but base classes often are also sufficient. One should never check or rely on subclassing of the object passed to method, because type declared in method signature, should be the final type we operate in scope of method. Applying to those straightforward rules allows us to create robust and polymorphic OOP design.

Other important feature is that Value Objects are immutable, i.e. they can not be modified after creation. Only valid way to create Value Object is to pass all required informations to constructor (and should be validated somewhere there). No setter methods should take place.

This post isn’t about obvious advantages of representing domain logic with support of Value Object. As well, we wouldn’t elaborate here about pros and cons of immutable objects. I’d rather would like to show an attempt to change Value Object, keeping it still immutable and using one of most bizarre, in my opinion, feature of PHP language, which is accessing private fields from outside an object.

Often times you would like to alter Value Object, by creating new one based on current (which is only valid way). Altering logic conceptually belongs to Value Object class, so should be located there. In such method, you clone current instance, set new information to given field and return the copy. And this is were accessing private fields for same class makes sense. You can do that without having additional setters which will break the design.

As illustrated above, all Value Object features are in place. This example is of course trivial, but you can imagine a lot more complicated VO’s, like values shared between bounded contexts.

First of all, we all need to admit, that quality of modern MVC framework raised a lot, comparing with state of things few years ago. Speaking about PHP – at the time, when I attracted my attention to this language, there were pure wilderness. We did not have any strong framework (unlike Ruby On Rails, which were sine qua non choice for Ruby web development). That caused multiple projects development, some of them are dead now (or should be), some hasn’t got good market adaptation and some of them are industry leaders at the moment (Symfony and Zend).

On the other hand, there’s huge temptation to write own frameworks, ignoring the great work of community. That has some advantages, in case you know exactly what you’re doing. Only one good reason for me is performance concerns. But still, doing everything by hand proves lack of understanding of tools and leads to giant problems with system maintenance. For me, it’s hard to imagine how one could create a complex system without usage of good framework. What’s more, its economically unreasonable to recreate the code, that already exists.

Alright, it’s clear –  applications which will serve their purpose are way easier to be created with modern framework. The choice isn’t easy (as well as choice of language), but if you ask me, I’ll say: pick the one you feel the most comfortable with and which is built on top of best design patterns. A framework won’t do the job by its own, though. And this is the point I’d like to make: don’t be bound to the framework. The best quote to reflect this point of view is:

The architecture of an accounting app should scream „accounting” not Spring & Hibernate.
Robert C. Martin via https://twitter.com/unclebobmartin/status/118365858581069824

By decoupling from framework (see Jakub Zalas slides) you’ll benefit in multiple ways: your code will be loosely coupled, easier to understand, readable, testable and most important: it will be robust. If for some reason, you’ll have to change framework (because yours isn’t supported any more and super 3rd edition of famous framework comes to general availability), you’ll spend considerably less amount of time to migrate to new libraries.

A thing to remember is, that good software design practices, such as design patterns or SOLID principles, exists for years now. They are applied in all software languages and you’ll find similar concepts both in Java Spring and PHP Symfony. Frameworks, on the other hand comes and goes. In 3 years from now, there won’t be Symfony2 or Zend Framework2, but your code will be still alive and need to be maintained. It’s your choice, if it be completely dependent on framework or if it will rely on proven patterns.

I strongly encourage to read and apply philosophy of Domain Driven Design. It’s better to focus on a core domain and reflect business needs by modelling them with code. Once you’ll be focused on domain, you’ll start to see that framework is only implementation detail and you’ll stop calling yourself Symfony Developer or Zend Developer, but rather Software Developer.

PS. LiipFunctionalTestBundle comes with similar concepts, maybe it’s worth having a look.

IsolatedTestsTrait is available here as a Gist.

For your convenience I’m also putting the code below. Feel free to comment and propose improvements!
[gist id=”7137331″]