# Distributed Publish/Subscribe Query Processing on the Spatio-Textual Data
Stream

Zhida Chen†, Gao Cong†, Zhenjie Zhang‡, Tom Z.J. Fu‡♯, Lisi Chen♮
$\;{}^{\dagger}$Nanyang Technological University, Singapore
$\;{}^{\ddagger}$Advanced Digital Sciences Center, Illinois at Singapore Pte.
Ltd., Singapore
$\;{}^{\sharp}$Guangdong University of Technology, China $\;{}^{\natural}$Hong
Kong Baptist University, Hong Kong
<EMAIL_ADDRESS><EMAIL_ADDRESS><EMAIL_ADDRESS><EMAIL_ADDRESS>
<EMAIL_ADDRESS>
###### Abstract

Huge amount of data with both space and text information, e.g., geo-tagged
tweets, is flooding on the Internet. Such spatio-textual data stream contains
valuable information for millions of users with various interests on different
keywords and locations. Publish/subscribe systems enable efficient and
effective information distribution by allowing users to register continuous
queries with both spatial and textual constraints. However, the explosive
growth of data scale and user base has posed challenges to the existing
centralized publish/subscribe systems for spatio-textual data streams.

In this paper, we propose our distributed publish/subscribe system, called _PS
2Stream_, which digests a massive spatio-textual data stream and directs the
stream to target users with registered interests. Compared with existing
systems, _PS 2Stream_ achieves a better workload distribution in terms of both
minimizing the total amount of workload and balancing the load of workers. To
achieve this, we propose a new workload distribution algorithm considering
both space and text properties of the data. Additionally, _PS 2Stream_
supports dynamic load adjustments to adapt to the change of the workload,
which makes _PS 2Stream_ adaptive. Extensive empirical evaluation, on
commercial cloud computing platform with real data, validates the superiority
of our system design and advantages of our techniques on system performance
improvement.

##  I Introduction

Driven by the proliferation of GPS-equipped mobile devices and social media
services, data containing both space and text information is being generated
at an unprecedented speed. Millions of social-media users, for example, are
uploading photos to Instagram with both location and text tags, posting geo-
tagged tweets on Twitter, and creating location-aware events in Facebook,
using their smart phones. Such user generated content arrives in a streaming
manner, and contains valuable information to both individual and business
users. On the one hand, individual users may be interested in events in
particular regions, and are keen to receive up-to-date messages and photos
that originate in the interested regions and are relevant to the events. On
the other hand, business users, e.g., Internet advertisers, expect to identify
potential customers with certain interest at a particular location, based on
their spatio-textual messages, e.g., restaurant diners in a target zone. For
both types of users, it is important to find spatio-textual messages
satisfying location and textual constraints in real-time, to deliver the
service for their business models.

Publish/subscribe systems [1, 2, 3, 4, 5, 6] provide the basic primitives to
support such information processing paradigms, such that _subscribers_
register subscription (continuous) queries in the system to catch all messages
from _publishers_ satisfying the query predicates. In the context of our
problem setting, each registered query from the subscribers consists of two
components, a space component describing the spatial region of interests and a
text component containing a boolean expression of keywords for matching. A
message matches a query, if its location lies in the interested region and its
textual content satisfies the boolean expression of the query. When a massive
spatio-textual data stream floods on the Internet, the publish/subscribe
system filters the messages and routes the matching messages to subscribers in
real-time.

Existing publish/subscribe systems are capable of handling subscriptions on a
spatio-textual data stream at moderate rates, by utilizing indexing techniques
tailored for spatio-textual data on a centralized server [7, 8, 9, 10, 11, 12,
13]. With the growth of spatio-textual messages on social media and registered
queries, the computation workload for publish/subscribe systems is quickly
increasing, which exceeds the capacity of a single server. This calls for a
distributed solution to building a publish/subscribe system over the spatio-
textual data stream. Moreover, due to the dynamic nature of the social media,
the workload of publish/subscribe systems varies dramatically over time. The
system is expected to conduct dynamic load adjustments with small migration
cost to fit for the change in the workload.

In this paper, we present our system, called PS2Stream, as a scalable
distributed system over a spatio-textual data stream, to enable real-time
publish/subscribe services on top of a cluster of servers. The system design
objective is to accomplish a maximal processing throughput, minimal latency,
and ignorable migration cost. To accomplish these goals, we propose a handful
of new techniques. Firstly, we propose a new workload partitioning strategy,
utilizing both text and space properties of the data, to facilitate
distributing the workload with aims of minimizing the total amount of workload
and load balancing of workers. Different from existing distributed systems, we
need to distribute both spatio-textual data stream and subscription query
stream, and different ways of workload distribution will result in different
amounts of workload. Secondly, we discuss dynamic load adjustment approaches
over our system architecture, in both local and global manners, to minimize
both the overhead of workload reassignment and the total amount of workload in
the scenario of workload changing. The main contributions of our work are
summarized as follows:

1) We propose a new workload partitioning algorithm that considers both space
and text properties of the data for the workload distribution. Our system is
the first to consider the optimal workload partitioning problem for
distributing both spatio-textual data stream and subscription query stream,
which aims at minimizing the total amount of workload, and balancing the load
of workers.

2) We propose efficient dynamic load adjustment algorithms to adjust the load
of workers in the scenario of workload changing. By considering both space and
text properties of the data, our dynamic load adjustment algorithms achieve
features of reducing the total amount of workload and invoking small migration
cost. To the best of our knowledge, our system is the first to support such
dynamic load adjustments.

3) We conduct extensive experiments on Amazon EC2 with a real-life spatio-
textual data stream. The results demonstrate that: i) our distribution
framework considering both space and text properties performs better than the
baselines utilizing either space or text property only, and our system
achieves excellent performance on both processing throughput and query
response latency; ii) our dynamic load adjustment approaches improve the
performance of the system and invoke small migration cost.

##  II Related Work

Spatial-Keyword Publish/Subscribe. The problem of building publish/subscribe
systems over a spatio-textual data stream on a centralized server has been
studied recently [7, 8, 9, 10, 11, 12, 13]. They focus on developing new
indexes to speed up the matching between spatio-textual objects and the
spatial-keyword continuous queries. Specifically, Chen et al [10] present an
indexing structure called IQ-tree. Li et al [7] propose an R-tree based
indexing structure. Wang et al [9] propose to adaptively group the continuous
queries using keyword partitions or space partitions based on a cost model.
The problem of similarity based spatial-keyword publish/subscribe is also
studied [8, 13], which determines the matching between a subscription and a
published spatio-textual object based on a similarity score.

In theory, these studies are complementary to our work as these complicated
index structures could be employed in workers of our system. However, it would
be expensive to jointly maintain and migrate these index structures across
multiple workers, which is required in our system.

Distributed Content based Publish/Subscribe. Our work is related to content
based subscription queries. Many works [1, 2, 3, 4, 5, 6] exist for
distributed content-based publish/subscribe systems over a wide-area network.
They aim at finding an optimal assignment of the subscribers to message
brokers with some performance criterion such as the latency [2]. Another line
of works[14, 15] consider deploying the publish/subscribe systems on a P2P
network, which focus on minimizing the communication cost. E-StreamHub [16] is
a distributed content based publish/subscribe system deployed on a cluster of
local servers. However, these systems do not provide supports for the spatio-
textual data, which is essential to our problem, and we do not see any
sensible way to deploy these systems for our problem.

Our system differs from these systems in two aspects. Firstly, our system is
based on the optimal workload partitioning problem, which aims at both of
minimizing the total amount of workload and balancing the load of workers in
distributing the workload of both subscription query stream and spatio-textual
data stream. Secondly, we support dynamic load adjustments by considering
minimizing both total amount of workload and migration cost.

Distributed Systems for Spatial Data. Many distributed systems [17, 18, 19,
20, 21] have been proposed for large-scale spatial data. Most of them use
existing spatial indexes to partition the data to different servers, e.g.,
SpatialHadoop [18] uses a grid index and a R-tree, and MD-HBase [20] uses a
kd-tree and a Quad tree. AQWA [21] uses a kd-tree to partition the spatial
data, which aims at minimizing the querying cost based on a query workload.
Our system differs from these systems in two ways. Firstly, we consider the
workload composed of updating highly dynamic subscription queries and
processing spatio-textual objects, while they consider processing disposable
queries on a static spatial data set. Secondly, our system design objectives
are different. We consider minimizing the total amount of workload and load
balancing of workers in distributing workload, as well as dynamic load
adjustments. We evaluate for the first time using the data partitioning
strategies in these systems for our problem in our experiments.

Distributed Data Stream Processing. The distributed data stream processing
systems are also related to our work. A common scheme is to split a continuous
query into multiple operators and distribute them to a cluster of servers.
NiagaraCQ [22] detects the commonality of different continuous queries and
combines their common operators. PSoup [23] improves the processing efficiency
of stateful operators such as join and aggregate. Subsequent systems like
Aurora*[24] and Borealis [25] further improve the scalability of the data
stream processing systems by allowing the same operator being executed by
multiple servers. These systems lack support for processing the spatio-textual
data stream and their proposed methods do not apply for our system.

A recent demonstration system, called Tornado [26], is presented for indexing
spatio-textual data streams to handle querying spatio-textual data streams.
However, it does not handle the stream of subscription queries. Furthermore,
unlike our system, Tornado does not consider the optimal workload partitioning
problem, and it does not support dynamic load adjustments which reduce the
total amount of workload with small migration cost. Tornado uses a kd-tree to
partition the workload, which has been included in our evaluation of the
baselines.

##  III Problem Statement and System Architecture

###  III-A Problem Statement

Spatio-Textual Object: A spatio-textual object is defined as $o=\langle
text,loc\rangle$, where $o.text$ is the textual content of object $o$ and
$o.loc$ is the geographical coordinate, i.e., latitude and longitude, of
object $o$.

In this work, we consider a stream of spatio-textual objects, such as geo-
tagged tweets, check-ins in Foursquare, geo-tagged posts in Instagram, etc. We
aim at building a distributed publish/subscribe system over a stream of
spatio-textual objects.

Users may express their interests on the spatio-textual objects with
subscription queries. Following the previous works [7, 9, 10], each
subscription query contains a Boolean keyword expression and a region. If a
new spatio-textual object falls in the specified region and satisfies the
Boolean keyword expression specified by a subscription query, the object will
be pushed to the user who submits the query. A subscription query is valid
until the user drops it. We next present the spatio-textual subscription
query.

Spatio-Textual Subscription (STS) Query: A Spatio-Textual Subscription (STS)
Query is defined as $q=\langle K,R\rangle$, where $q.K$ is a set of query
keywords connected by AND or OR operators, and $q.R$ denotes a rectangle
region.

A spatio-textual object $o$ is a result of an STS query $q$ if $o.text$
satisfies the boolean expression of $q.K$ and $o.loc$ locates inside $q.R$.

The large number of STS queries and high arrival rate of spatio-textual
objects call for a distributed solution. We build our system on a cluster of
servers with several servers playing the role of dispatchers, which distribute
the workload to other servers. The workload to our system includes the
insertions and deletions of STS queries, and the matching operations between
STS queries and spatio-textual objects.

Problem Statement We aim at building a distributed publish/subscribe system
over a spatio-textual data stream. We expect the system to have the features
that (i) the system can achieve a high throughput, (ii) each tuple can be
processed in real-time, i.e., low latency, and (iii) the system can
dynamically adjust the load of servers when the workload changes.

###  III-B System Architecture

We present a distributed Publish/Subscribe system for handling subscription
queries over a Spatio-textual data stream, and we call the system PS2Stream.
There are three components in PS2Stream: dispatcher, worker and merger, which
are shown in Figure 1. The dispatcher takes as input a stream of spatio-
textual objects, and it receives two types of requests from users, namely
submitting new subscriptions and dropping existing subscriptions, which
correspond to two operations, i.e., query insertions and deletions,
respectively. The dispatcher assumes the responsibility of distributing
spatio-textual objects and query insertions/deletions. The worker accepts the
workload sent from the dispatcher and conducts the following operations
accordingly:

(1) _Query insertion_ : On receiving a new STS query, the worker inserts it
into an in-memory index maintained in the worker;

(2) _Query deletion_ : On receiving the indication of deleting an existing STS
query, the worker removes the query from the index;

(3) _Matching a spatio-textual object_ : On receiving a spatio-textual object,
the worker checks whether the object can be a match for any STS query stored
in the worker. If yes, the matching result is forwarded to the merger.

The merger assumes the responsibilities of removing duplicated matching
results produced by workers and sending the results to corresponding users.

Figure 1: Architecture of PS2Stream

##  IV Hybrid Partitioning

In this section, we present the design of our PS2Stream system. We define a
new workload distribution problem in Section IV-A, and provide a solution to
it in Section IV-B. We introduce the index structure adopted in dispatchers in
Section IV-C. Finally, we discuss how workers process their workload in
Section IV-D.

###  IV-A Definition

We propose PS2Stream for processing a stream of subscription queries over a
spatio-textual data stream. Different from most existing distributed systems,
we need to distribute the workload of both updating highly dynamic STS queries
and processing spatio-textual objects. In our problem, different ways of
workload distribution will result in different amounts of workload. During the
workload distribution, the following two factors are important for the system.
One is the total amount of workload distributed to the workers, and the other
one is the load balancing of workers.

Next, we first define the load of one worker. Based on it, we propose our
Optimal Workload Partitioning problem. To the best of our knowledge, our
definition of the workload distribution problem is the first for distributing
spatio-textual data that considers minimizing the total amount of workload
while balancing the load of workers.

Figure 2: Two regions with different data distributions.

###### Definition 1

Load of One Worker: Given a time period, the load of a worker $w_{i}$ for the
period is computed by

$L_{i}=c_{1}\cdot|O_{i}|\cdot|Q_{i}^{i}|+c_{2}\cdot|O_{i}|+c_{3}\cdot|Q_{i}^{i}|+c_{4}\cdot|Q_{i}^{d}|,$

where $O_{i}$ is the set of spatio-textual objects sent to $w_{i}$,
$Q_{i}^{i}$ is the set of STS query insertion requests sent to $w_{i}$, and
$Q_{i}^{d}$ is the set of STS query deletion requests sent to $w_{i}$ in the
given time period. Here $c_{1}$ is the average cost of checking whether a
spatio-textual object matches an STS query, $c_{2}$ is the average cost of
handling one object, $c_{3}$ is the average cost of handling one STS query
insertion request, and $c_{4}$ is the average cost of handling one STS query
deletion request.

###### Definition 2

Optimal Workload Partitioning Problem: Given a set of spatio-textual objects
$O$, a set of STS query insertion requests $Q^{i}$ and a set of STS query
deletion requests $Q^{d}$, let $S$ denote the space they reside in and $T$
denote the set of terms appearing in them. We aim to partition $S$ into $m$
subspaces $S_{1},S_{2},\cdots,S_{m}$ and partition $T$ into $m$ subsets
$T_{1},T_{2},\cdots,T_{m}$, and assign each pair $(S_{i},T_{i})$ to one
worker, where $m$ is the number of workers. An object $o$ is sent to worker
$w_{i}$ only if $o.loc\in S_{i}$ and $o.text\cap T_{i}\neq\emptyset$. An
insertion/deletion request of an STS query $q$ is sent to worker $w_{i}$ only
if $q.R\cap S_{i}\neq\emptyset$ and $q.K\cap T_{i}\neq\emptyset$. The Optimal
Workload Partitioning problem is to find a partition such that
$\sum_{i=1}^{m}L_{i}$ is minimized, subject to the constraint that
$\forall{i\neq j}$ and $L_{i}\geq L_{j},L_{i}/L_{j}\leq\sigma$, where $\sigma$
is a small constant value being larger than 1.

Note that the constraint in the problem statement is to achieve load
balancing.

###### Theorem 1

The Optimal Workload Partitioning problem is NP-hard.

Proof: We prove by reducing from the Partition problem, which is NP-hard. In
the Partition problem, given $n$ numbers
$a_{1},a_{2},\cdots,a_{n}\in\mathbf{N}$, we decide whether there exists a set
$S\subseteq\\{1,2,\cdots,n\\}$ such that $\sum_{i\in S}a_{i}=\sum_{i\notin
S}a_{i}$. For any instance of the Partition problem, we can reduce it to one
instance of the Optimal Workload Partitioning Problem in the following way:

First, we set the value of $m$ to 2 and set the value of $\sigma$ to 1, which
means that there are 2 workers in the system and we want them to be perfectly
load balancing.

Second, we produce the workload of processing the spatio-textual objects only
($Q^{i}=\emptyset$ and $Q^{d}=\emptyset$). The objects share the same textual
content and are distributed to $n$ non-overlapping regions
$r_{1},r_{2},\cdots,r_{n}$. Let $O_{i}$ denote the set of objects locating in
region $r_{i}$ and $|O_{i}|=a_{i}$.

Since all objects share the same textual content, to satisfy the load
balancing constraint, we cannot partition them by the text property, i.e., the
complete term set $T$ is assigned to all workers. Hence the solution is to
find a 2-partition $R_{1},R_{2}$ of $\\{r_{1},r_{2},\cdots,r_{n}\\}$
satisfying that $\sum_{r_{i}\in R_{1}}|O_{i}|=\sum_{r_{j}\in R_{2}}|O_{j}|$.
It is equivalent to the Partition problem, thus completing our proof.

###  IV-B Workload Partitioning

Due to the hardness of the Optimal Workload Partitioning problem, we
investigate heuristic algorithms to partition the workload. The text property
and the space property of the data inspire us to partition the workload by the
text property or the space property. Previous works [27, 28] on constructing a
distributed text retrieval system or a distributed information dissemination
system use text-partitioning to partition the workload. However, they focus on
the communication cost and load balancing. Other works on constructing a
distributed system for a static spatial data set [17, 18, 19, 20, 21] and for
a spatio-textual data stream [26], use space-partitioning to partition the
workload, and they do not consider minimizing the total amount of workload and
the load balancing collectively.

Such partitioning algorithms based on an unanimous scheme perform poorly in
minimizing the total amount of workload when the data distributions among
different regions are quite different. As shown in Figure 2, space-
partitioning performs well in region $r_{2}$ where the objects and queries are
well spread, but it does not fit for region $r_{1}$ as the ranges of queries
in $r_{1}$ are large and clustered. Similarly, text-partitioning is good for
$r_{1}$ but bad for $r_{2}$.

To better solve the Optimal Workload Partitioning problem, we propose a new
partitioning algorithm that utilizes both text and space properties of the
data. Our hybrid partitioning algorithm decomposes the workload into a set of
units by wisely using the text property or the space property in different
regions, with the goal of minimizing the total amount of workload after the
decomposition. Additionally, we distribute those units to workers, so that the
load balancing constraint can be satisfied.

Algorithm overview. The core idea of our algorithm is to first use space-
partitioning to identify the subspaces that have large differences in the text
distribution between objects and queries, and then for each subspace, we
choose between space-partitioning and text-partitioning to minimize the total
amount of workload.

The algorithm can be divided into two phases. In the first phase, the
algorithm divides the space into subspaces based on the text similarity
between objects and queries. The purpose is to identify those subspaces where
the text-partitioning would perform better. To achieve this, for a subspace
where the text similarity between objects and queries is smaller than a
threshold, we check whether partitioning it will result in new subspace(s)
where the text similarity becomes smaller. If yes, we partition the subspace
recursively and perform the same checking for each new subspace. At the end of
this phase, we obtain two types of subspaces represented by $N_{s}$ and
$N_{t}$, respectively. The subspaces in $N_{t}$ have small text similarity
between objects and queries, and we partition them using text-partitioning
only in the second phase. For the subspaces in $N_{s}$, we compare the
workloads produced by space-partitioning and text-partitioning, respectively,
and select the strategy resulting in a smaller workload. This comparison is
necessary as when the query ranges are very large, using space-partitioning
will invoke queries being duplicated to multiple workers, which may result in
a larger workload than using text-partitioning even though the objects and
queries have large text similarity. In the second phase, if the number of
partitions is smaller than the number of workers, we further partition the
workload in $N_{s}$ and $N_{t}$. To minimize the total amount of workload, we
design a dynamic programming function to find the optimal number of partitions
for each subspace in $N_{s}$ and $N_{t}$. We then check whether the load
balancing constraint can be satisfied. If no, we recursively further partition
one subspace until the load balancing constraint can be satisfied.

The output of our workload partitioning procedure is an index structure named
kdt-tree, which is a kd-tree with some leaf nodes being further partitioned by
the text property. Figure 3 shows an example of a kdt-tree, which distributes
the workload to 4 workers: the set of leaf nodes is divided into 4 disjoint
subsets and each subset is assigned to one worker randomly, e.g.,
$\\{N_{11}\\}$ is assigned to worker $w_{1}$ and $\\{N_{12},N_{31}\\}$ is
assigned to worker $w_{2}$. For an object and query locating inside or
overlapping with the space range of node $N_{2}$ or $N_{4}$, it is sent to
worker $w_{3}$ or $w_{4}$ without checking the textual content. When an object
or query locates inside or overlaps with the space range of node $N_{1}$, its
textual content is checked and it is sent to worker $w_{1}$ (resp. $w_{2}$) if
it contains terms in $T_{1}$ (resp. $T_{2}$). Workers $w_{2}$, $w_{3}$, and
$w_{4}$ also receive objects and queries that locate inside or overlap with
the space range of node $N_{3}$ and contain terms in $T_{1}^{{}^{\prime}}$,
$T_{2}^{{}^{\prime}}$ and $T_{3}^{{}^{\prime}}$, respectively.

Figure 3: Example of a kdt-tree

Input: The number of workers $m$, Load balance constraint threshold $\sigma$,
Text similarity threshold $\delta$, A set of objects $O$, A set of queries
$Q$;

Output: The root node of a kdt-tree with each leaf node assigned a number in
$\\{1,2,\cdots,m\\}$

1

2

3Initialize the kdt-tree structure with a root node $n_{r}(O,Q)$;

4 $N_{u}\leftarrow\\{n_{r}\\}$, $N_{t}\leftarrow\emptyset$,
$N_{s}\leftarrow\emptyset$;

5 while  _$|N_{u}| >0$_ do

6 Pop one node $n$ from $N_{u}$;

7 if  _$sim_{t}(O_{n},Q_{n})\geq\delta$_ then

8 $N_{s}\leftarrow N_{s}\cup\\{n\\}$;

9

10 else

11 $\\{n_{1},n_{2}\\}\leftarrow$ partition $n$ into 2 nodes in the direction
that minimizes
$\alpha=\textrm{min}\\{sim_{t}(O_{n_{1}},Q_{n_{1}}),sim_{t}(O_{n_{2}},Q_{n_{2}})\\}$;

12

13 if  _$|\alpha-sim_{t}(O_{n},Q_{n})|\approx 0$_ then

14 $N_{t}\leftarrow N_{t}\cup\\{n\\}$;

15

16 else

17 $N_{u}\leftarrow N_{u}\cup\\{n_{1},n_{2}\\}$;

18

19

20

21if  _$|N_{t}|+|N_{s}| <m$_ then

22 $A\leftarrow\textnormal{{ComputeNumberPartitions}}(N_{t},N_{s},m)$;

23 for  _each node $n$ in $N_{t}\cup N_{s}$_ do

24 PartitionNode ($n$, $N_{t}$, $N_{s}$, $A[n]$);

25

26

27while  _True_ do

28 MergeNodesIntoPartitions ($N_{t}$, $N_{s}$, $m$);

29 $L_{max}\leftarrow$ the maximum load value among all partitions;

30 $L_{min}\leftarrow$ the minimum load value among all partitions;

31 if  _$L_{max}/L_{min}\leq\sigma$_ then

32 break;

33

34 else

35 $n\leftarrow$ the node having the largest load value in $N_{t}\cup N_{s}$;

36 PartitionNode ($n$, $N_{t}$, $N_{s}$, 2);

37 if  _$|N_{t}|+|N_{s}|\geq\theta$_ then

38 break;

39

40

41

42

43return $n_{r}$;

Algorithm 1 Partition Workload

Algorithm 1 presents the pseudo code of our workload partitioning algorithm.
The algorithm first initializes a root node $n_{r}(O,Q)$ of the kdt-tree and
puts it into $N_{u}$ (lines 1–2). The variable $N_{u}$ denotes the set of
nodes that have not been decided to put into $N_{t}$ or $N_{s}$. A while loop
is conducted to compute $N_{t}$ and $N_{s}$ (lines 3–12). At each iteration,
it pops one node $n$ from $N_{u}$ (line 4) and computes the text similarity
between the objects and queries in $n$. We use cosine similarity in our
algorithm. If the text similarity is larger than a threshold $\delta$, we
consider node $n$ as being not suitable for text-partitioning, and add it into
$N_{s}$ (line 6). Otherwise, we split node $n$ in either x-dimension or
y-dimension as the normal kd-tree does, except that we prefer the direction
resulting in a smaller text similarity $\alpha$ between objects and queries in
the new subspaces (line 8). If the difference between $sim_{t}(O_{n},Q_{n})$
and $\alpha$ is minor, we consider node $n$ as being consistent in text
similarity of the objects and queries in it and add $n$ into $N_{t}$ (lines
9–10). Otherwise, we add the new nodes $n_{1}$ and $n_{2}$ into $N_{u}$ (line
12). In the second phase, the algorithm proceeds to check the number of nodes
in $N_{t}$ and $N_{s}$ (line 13). If the number of nodes is smaller than $m$,
i.e., the number of workers, a dynamic programming function
ComputeNumberPartitions is called to compute the number of partitions for each
node (lines 13–14). It is for minimizing the total amount of workload by
determining an optimal number of partitions for each node. The function
PartitionNode is then called to partition the nodes (lines 15–16). After that,
the algorithm recursively checks whether the load balancing constraint can be
satisfied (lines 18–22) and partitions the node having the largest load, until
the load balancing constraint is satisfied or the number of nodes reaches a
threshold $\theta$, where $\theta$ is a threshold of the maximum number of
nodes (lines 24–27). In the end, the algorithm returns the root node of the
kdt-tree (line 28).

Computing the number of partitions. For the purpose of minimizing the total
amount of workload, a dynamic programming function ComputeNumberPartitions is
called to compute the number of partitions for each node in $N_{t}$ and
$N_{s}$ when $|N_{t}|+|N_{s}|<m$ (lines 13–14 of Algorithm 1). The core of the
function is updating $L[i,j]$, where $L[i,j]$ denotes the minimum sum of loads
after partitioning the first $i$ nodes into a total number of $j$ partitions.
Let $C[i,k]$ denote the sum of loads after partitioning the $i$th node into
$k$ partitions. The value of $C[i,k]$ can be obtained by calling a function
similar to PartitionNode except that the node is not really partitioned. Then
we can compute $L[i,j]$ by using the equation

$L[i,j]\leftarrow min_{1\leq k\leq j-i+1}\\{L[i-1,j-k]+C[i,k]\\}.$

Based on the arrays $L$ and $C$, the function can construct a map which maps
from each node to the corresponding number of partitions.

Node partitioning. In Algorithm 1, we need to partition a node, if the number
of nodes is smaller than the number of workers (lines 13–16), or the load
balancing constraint cannot be satisfied (lines 23–27). Function PartitionNode
takes as input the node $n$ to be split, the set of nodes $N_{t}$, the set of
nodes $N_{s}$, and the number of splits $p$. It outputs $p$ new nodes by
partitioning node $n$. If node $n$ belongs to $N_{t}$, it indicates that the
text similarity between objects and queries in $n$ is small, and we partition
$n$ into $p$ splits using text-partitioning. Otherwise, we compare the
workloads produced by using text-partitioning or space-partitioning to
partition $n$, and select the strategy resulting in a smaller workload.

Node merging. When the number of leaf nodes in the kdt-tree exceeds the number
of workers $m$, we call function MergeNodesIntoPartitions to divide those leaf
nodes into $m$ partitions (line 18 of Algorithm 1). The function first sorts
the leaf nodes in descending order of their loads. In that order, for each
leaf node $n$, it finds the partition $part$ such that adding $n$ to $part$
will result in a minimum load increase. If assigning $n$ to $part$ does not
increase the load balancing factor (i.e., $L_{max}/L_{min}$), $n$ is assigned
to $part$. Otherwise, $n$ is assigned to the partition that has currently the
smallest load.

###  IV-C Index Structure on Dispatchers

The kdt-tree can be used in the dispatcher to distribute the workload. For
each spatio-textual object or updating request of an STS queries, the
dispatcher obtains the corresponding leaf node(s) of the kdt-tree by
traversing from the root node. The procedure takes $O(\log(m))$ time, which
may overload the dispatcher when the arrival speeds of objects and updating
requests of queries are very fast.

To alleviate the burden of the dispatcher, instead of maintaining a kdt-tree,
we conduct the workload distribution using a gridt index with each grid cell
containing two hash maps $H_{1}$ and $H_{2}$. $H_{1}$ maps from terms in the
complete term set $T$ to the worker ids, and $H_{2}$ maps from terms in STS
queries to the worker ids. To distribute a spatio-textual object, the
dispatcher first obtains the cell containing the object, and uses $H_{2}$ to
find the destination worker(s). The object can be discarded if it contains no
terms in $H_{2}$. To distribute an updating request of an STS query (note that
the request contains complete information of the STS query), if the query
keywords are connected by AND operators only, the dispatcher obtains the cells
that query overlaps with, and looks up $H_{1}$ using the least frequent
keyword in each obtained cell to find the destination worker(s). $H_{2}$ is
updated correspondingly if it does not contain the keyword. For the query
containing OR operators, similar operations are conducted except that the set
of the least frequent keywords in each conjunctive norm form are used to find
the destination worker(s). The gridt index can be built from the kdt-tree. The
granularity of the cell is decided by the leaf nodes of the kdt-tree. Figure 4
shows an example of the relation between the gridt index and the kdt-tree in
Figure 3.

Figure 4: A gridt index transformed from a kdt-tree

###  IV-D Query Processing on Workers

Each worker maintains an in-memory index structure to organize the STS queries
for accelerating the matching of incoming spatio-textual objects. The index
should be efficient in both matching STS queries and spatio-textual objects
and updating of STS queries. In our system, we adopt an index structure named
Grid-Inverted-Index (or GI2 for short) [29]. We choose GI2 due to its
efficiency in construction and maintaining, which is important for processing
a dynamic workload like the data stream. Note that our system can be extended
to adopt other index structures.

After dividing the STS queries by the cells they overlap with, GI2 constructs
an inverted index for each cell to organize the STS queries. For the query
containing AND operators only, it is appended to the inverted list of the
least frequent keyword. For the query containing OR operators, it is appended
to the inverted lists of the least frequent keywords in each conjunctive norm
form. The granularity of the cells can be determined empirically in
experiments.

When processing a spatio-textual object $o$, the worker first looks up which
cell contains $o$ and then checks the associated inverted list in the cell for
each distinct term in $o.text$ to see whether $o$ can be matched by any STS
query $q_{i}$ in that list. If there is a match, object $o$ is a result of
$q_{i}$.

To delete an STS query from GI2, we adopt the lazy-deletion strategy.
Specifically, we do not delete an STS query immediately after receiving the
deletion request. Instead, we record the ids of the queries to be deleted into
a hash table and remove the queries during the object matching procedure. In
particular, while traversing an inverted list, we check whether the id of each
query in the inverted list appears in the hash table. If yes, the query can be
deleted.

##  V Dynamic Load Adjustment

The dynamic property of the data stream gives rise to a changing workload. The
load of workers may change gradually over time. To handle this, we conduct two
types of load adjustments: local load adjustment and global load adjustment.

###  V-A Local Load Adjustment

When the dispatcher detects that the load balancing constraint is violated, it
notifies the most loaded worker $w_{o}$ to transfer part of its workload to
the least loaded worker $w_{l}$. We expect the load adjustment procedure
having the following two features: 1.) The total amount of workload can be
reduced after the load adjustment. 2.) The migration cost is small so that the
load adjustment can be conducted efficiently. Existing systems [30, 26, 28,
16] do not support dynamic load adjustments that consider both features. To
achieve the goal, we propose new load adjustment algorithms.

Algorithm overview. We adjust the workload by migrating STS queries. The
queries are migrated in the unit of one cell in the gridt index. Our load
adjustment algorithm is composed of two phases. In the first phase, we check
whether some cells in $w_{o}$ and $w_{l}$ can be split or merged so that the
total amount of workload can be reduced. We conduct related migration
operations if such cells exist. In the second phase, if the load balancing
constraint is still violated, we continue to select a set of cells in $w_{o}$
to be migrated to $w_{l}$, with the goals of minimizing the migration cost and
the load balancing constraint can be satisfied.

Phase I. We first check the $p$ most loaded cells $G_{p}$ in $w_{o}$, where
$p$ is a small parameter. For each cell $g_{s}$ not using text-partitioning in
$G_{p}$, we do the following checking: after using text-partitioning to
partition $g_{s}$ into two new cells $g_{1}$ and $g_{2}$, whether the total
amount of workload can be reduced if $g_{1}$ or $g_{2}$ is migrated to
$w_{l}$. If yes, we conduct the text-partitioning on $g_{s}$ and migrate the
cell having a smaller size between $g_{1}$ and $g_{2}$ to $w_{l}$. For each
cell $g_{t}$ using text-partitioning in $G_{p}$, if there exists a cell
$g_{t}^{{}^{\prime}}$ in $w_{l}$ which shares the same space region as $g_{t}$
has, we check whether migrating $g_{t}$ to $w_{l}$ and merging $g_{t}$ and
$g_{t}^{{}^{\prime}}$ can reduce the total load. If yes, we conduct the
migration.

Phase II. If the load balancing constraint is still violated, we conduct the
second phase, which solves the Minimum Cost Migration problem. Definition 3
defines the load of a cell, and Definition 4 presents the definition of the
Minimum Cost Migration problem.

###### Definition 3

Given a time period, the load of a cell $g$ for the period is computed by

$L_{g}=n_{o}\cdot n_{q},$

where $n_{o}$ is the number of spatio-textual objects falling in cell $g$ and
$n_{q}$ is the average number of STS queries stored in cell $g$ in the given
time period.

###### Definition 4

Minimum Cost Migration: Let $\tau$ denote the amount of load to be migrated.
Consider a worker $w_{o}$ with the set of cells $G_{o}$. The minimum cost
migration problem is to compute a set of cells $G_{s}$ to be migrated from
worker $w_{o}$.

$G_{s}=\underset{G_{s}}{\arg\min}\ \ \sum_{g\in G_{s}}S_{g}\ \ \textrm{s.t. }$
$\sum_{g\in{G_{s}}}L_{g}\geq\tau,$

where $S_{g}$ is the total size of the queries in cell $g$.

###### Theorem 2

The Minimum Cost Migration problem is NP-hard.

Proof: We prove by reducing from the Partition Problem, which is NP-hard. In
the Partition problem, given $n$ numbers
$a_{1},a_{2},\cdots,a_{n}\in\mathbf{N}$, we decide whether there exists a set
$S\subseteq\\{1,2,\cdots,n\\}$ such that $\sum_{i\in S}a_{i}=\sum_{i\notin
S}a_{i}$. Let MCMP($X$, $Y$) denote the decision version of the Minimum Cost
Migration problem, and the solution to it returns “True” if there exists such
subset of cells $G_{s}$ satisfying that $\sum_{g\in{G_{s}}}L_{g}\geq X$ and
$\sum_{g\in G_{s}}S_{g}=Y$, and returns “False” otherwise. Given an arbitrary
instance of the Partition problem, we create an instance of MCMP($X$, $Y$) as
follows:

First, we produce the workload that for each cell $g_{i}$, $n_{o}=1$ and
$n_{q}=a_{i}$ (i.e., $L_{g_{i}}=a_{i}$), and $S_{g}=a_{i}$.

Second, we set $X=Y=\frac{\sum_{i=1}^{n}a_{i}}{2}$.

Therefore, the solution returns “True” iff there exists a set
$S\subseteq\\{1,2,\cdots,n\\}$ such that $\sum_{i\in S}a_{i}=\sum_{i\notin
S}a_{i}$, and returns “False” otherwise.

####  V-A1 Dynamic Programming Algorithm

We develop a dynamic programming algorithm to solve the Minimum Cost Migration
problem. Let $H(i,j)$ denote a subset of cells
$\\{g_{1},g_{2},\cdots,g_{i}\\}$ such that the total size of its cells is no
larger than $j$ and the total load is maximum

$H(i,j)=\underset{H\subseteq\\{g_{1},g_{2},\cdots,g_{i}\\}\land\sum_{g\in
H}S_{g}\leq j}{\arg\max}\sum_{g\in H}L_{g}.$

Let $A(i,j)$ denote the load of $H(i,j)$. We compute $A(i,j)$ for all
$i\in\\{1,2,\cdots,n\\}$ and $j\in(0,P]$, where $P$ is an upper bound of the
minimum migration cost. Following shows the computation of $A(i,j)$.

$A(i,j)=\begin{cases}A(i-1,j)&\text{ if }j\leq S_{g_{i}}\\\ max\\{A(i-1,j),\\\
\qquad A(i-1,j-S_{g_{i}})+L_{g_{i}}\\}&\text{ otherwise }\end{cases}.$

The time complexity of the dynamic programming algorithm is $O(nP)$ and it
takes long running time when the size of cells is large. Moreover, it needs
$O(nP)$ memory space, which is expensive. To address these issues, we next
propose Algorithm GR, which is more efficient and requires much less memory
space.

####  V-A2 Greedy Algorithm

We introduce the basic idea of the proposed greedy algorithm, denoted by GR.
For each cell $g$, we compute $S_{g}/L_{g}$, which indicates the relative cost
of migrating the cell. It is expected that migrating a cell with a smaller
relative cost may have smaller migration cost. We scan cells in $G_{o}$ in
ascending order of their relative costs to find a set of cells to be migrated.
For each cell we scan, if the total migrated load after including it in the
result is still less than $\tau$, the cell will be included in the result, and
we mark it by “GS”; otherwise, the cell becomes a candidate cell to be
included in the result, and we mark it by “GL”. The candidate cell will be
included in the result, if in the subsequent scan we cannot find cells that
can meet the load requirement and incur smaller migration cost.

We illustrate this process with Figure 5, where the cells are sorted by their
relative costs, and they are marked either by “GS” or “GL”. For any value of
$t$, we have $\sum_{i=1}^{t}\sum_{g\in GS_{i}}L_{g}<\tau$; and $\forall
g^{{}^{\prime}}\in GL_{t},\sum_{i=1}^{t}\sum_{g\in
GS_{i}}L_{g}+L_{g^{{}^{\prime}}}\geq\tau$. For any value of $t$, the set
$GS_{1}\cup GS_{2}\cup\dots\cup GS_{t}\cup\\{g^{\prime}\\}$ is a candidate
solution to the Minimum Cost Migration problem.

Figure 5: Greedy algorithm for computing the set of cells to be migrated

During the scan, we find the value for $t$ and the cell $g^{\prime}$ such that
$GS_{1}\cup GS_{2}\cup\dots\cup GS_{t}\cup\\{g^{\prime}\\}$ has the minimum
migration cost among all the candidate solutions that we have scanned.

###  V-B Global Load Adjustment

The performance of current workload partitioning strategy will degrade when
there exists a significant change in the data distribution. To handle this, we
periodically check whether a workload repartitioning is necessary on a recent
sample of data. If yes, we conduct a workload repartitioning using the
algorithm presented in Section IV-B. To avoid the large migration cost after
the workload repartitioning, we make a temporary compromise on the system
performance by maintaining two workload distribution strategies: one for the
old STS queries before the workload repartitioning, and the other one for the
new STS queries. When the amount of old STS queries becomes small, we conduct
the migration and stop the old workload distribution strategy. We set a long
period for the checking, e.g., once per day, which is reasonable as the data
distribution usually changes slowly.

##  VI Experimental Evaluation

###  VI-A Experimental Setup

We build PS2Stream on top of Storm111http://storm.apache.org, an open-source
stream processing engine. Our system is deployed on the Amazon EC2 platform
using a cluster of 32 c3.large instances. Each c3.large instance has 2 vCPUs
running Intel Xeon E5-2680 at 2.5GHz and 3.75 GB RAM. The network performance
is moderate and no further information about the bandwidth is provided by
Amazon. We rent another high performance m4.2xlarge instance (which is
equipped with 8 vCPUs and 32 GB RAM, and has high IO performance) to emit
streams of spatio-textual objects and STS queries to our system.

Datasets and STS Queries. Our experiments are conducted on two datasets, which
are TWEETS-US and TWEETS-UK. TWEETS-US consists of 280 million spatio-textual
tweets in America and TWEETS-UK consists of 58 million spatio-textual tweets
in Britain. Since we do not have real-life STS queries, we synthesize queries
based on tweets. The number of keywords is a random number ranging from 1 to 3
and the keywords are connected by either AND or OR operators. The query range
is a square and the center is randomly selected from the locations of tweets
in TWEETS-US or TWEETS-UK. We synthesize four groups of STS queries as
follows.

(1) STS-US-Q1, STS-US-Q2: The two groups of queries are synthesized based on
TWEETS-US. For STS-US-Q1, the side lengths of the rectangle are randomly
assigned between 1km and 50km and the keywords share the same distribution as
the terms in TWEETS-US, i.e., the keywords in queries satisfy the power-law
distribution. For STS-US-Q2, the side lengths of the rectangle are randomly
assigned between 1km and 100km and the queries contain at least one keyword
that is not in the top 1% most frequent terms in TWEETS-US.

(2) STS-UK-Q1, STS-UK-Q2: The two groups of queries are synthesized based on
TWEETS-UK as we do for TWEETS-US.

Workload. The ratio of processing a spatio-textual tweet to inserting or
deleting an STS query is approximately 5. The arrival speeds of requests for
inserting an STS query and deleting an STS query are equivalent. It indicates
that the number of STS queries in the system will be stable eventually. We use
a parameter $\mu$ to control the number of STS queries. We achieve this by
using a Gaussian distribution $\mathcal{N}(\mu,\sigma^{2})$ to determine the
number of newly arrived STS queries between inserting an STS query and
deleting it. We set different values of $\mu$ in the experiments with
$\sigma=0.2\mu$.

###  VI-B Evaluating Baselines

We evaluate the performance of baseline workload distribution algorithms.

Text-Partitioning. Text-partitioning algorithms divide the lexicon into
several partitions, each of which is assigned to one worker, and distribute
the workload based on the textual content of objects/queries. We implement the
following three text-partitioning algorithms: (1) Algorithm frequency-based
partitioning uses the frequency values of the terms to do the partitioning.
(2) Algorithm hypergraph-based partitioning [27] constructs a hypergraph based
on the co-occurrence of terms and partitions that hypergraph. (3) Algorithm
metric-based partitioning [28] uses a metric function to do the partitioning.

Space-Partitioning. Space-partitioning algorithms divide the space into
several partitions, each of which is assigned to one worker, and distribute
the workload based on the spatial content of objects/queries. We implement the
following three space-partitioning algorithms: (1) Algorithm grid partitioning
[18] represents the space as a set of cells and then partitions that set of
cells. We set its granularity as $2^{6}\times 2^{6}$, as it performs best
under that granularity. (2) Algorithm kd-tree partitioning [21, 26] constructs
a kd-tree to do the partitioning, of which each leaf node represents one
partition. We transform the kd-tree to a grid index to accelerate the workload
distribution in the dispatchers. (3) Algorithm R-tree partitioning [18]
constructs a R-tree to do the partitioning, and then partitions the set of
leaf nodes.

Note that this is the first work on empirically comparing the performance of
these algorithms for handling a spatio-textual data stream. These experiments
are conducted using 4 dispatchers and 8 workers. The granularity of GI2 index
of workers in all algorithms is set as $2^{6}\times 2^{6}$ for fairness.

(a) Text-Partitioning (#Q1=5M)

(b) Text-Partitioning (#Q2=10M)

(c) Space-Partitioning (#Q1=5M)

(d) Space-Partitioning (#Q2=10M)

Figure 6: Throughput Comparison (Baselines)

Throughput The throughput is the average number of tuples (i.e., including
both query insertion/deletion and object matching) being processed per second
when the processing capacity of the system is reached (i.e., the input speed
of the data stream is tuned to be closed to the maximum number of tuples that
the system can handle in unit time). We set $\mu=5$M for Q1 queries, and set
$\mu=10$M for Q2 queries.

Figure 6 shows the throughputs of the baselines. For Q1 queries, space-
partitioning algorithms perform better than text-partitioning algorithms. For
example, for STS-US-Q1, the throughput of grid partitioning is 160,659
tuples/second, and the throughput of frequency-based partitioning is 76,921
tuples/second. The reason is that space-partitioning algorithms impose a
smaller amount of workload on the workers: the text-partitioning algorithms
send each object to multiple workers as the keywords in Q1 queries are
frequent in objects. For Q2 queries, text-partitioning algorithms perform
better than space-partitioning algorithms. For example, for STS-US-Q2, the
throughput of metric-based partitioning is 267,415 tuples/second, and the
throughput of R-tree partitioning is 181,542 tuples/second. The reason is that
the Q2 queries have larger query ranges and less frequent keywords, and
therefore text-partitioning algorithms impose a smaller workload on the
workers.

Overall, metric-based partitioning performs the best among text-partitioning
algorithms, and kd-tree partitioning performs the best among space-
partitioning algorithms. Therefore, we select metric-based partitioning and
kd-tree partitioning for further evaluation.

###  VI-C Evaluating Hybrid Partitioning

We evaluate the performance of our hybrid partitioning algorithm in this set
of experiments. To simulate the situation that users in different regions have
different preferences over the spatio-textual objects, we synthesize two new
groups of STS queries: STS-US-Q3 and STS-UK-Q3. STS-US-Q3 is synthesized by
partitioning the spatial range of the USA into 100 regions of equal size, and
for each region we use STS-US-Q1 or STS-US-Q2. Similar procedure is conducted
to synthesize STS-UK-Q3

We first evaluate hybrid partitioning using 4 dispatchers and 8 workers. Then
we evaluate the scalability of hybrid partitioning using 4 dispatchers while
varying the number of workers.

(a) #Queries=5M (Q1)

(b) #Queries=10M (Q2)

(c) #Queries=10M (Q3)

Figure 7: Throughput Comparison

Throughput Figure 7 shows the throughputs of hybrid partitioning, metric-based
partitioning and kd-tree partitioning. Algorithm hybrid partitioning shows the
overall best performance regardless of the data distributions. In Figure 7(c),
hybrid partitioning outperforms metric-based partitioning and kd-tree
partitioning by 30% in terms of throughput. In Figure 7(a), for STS-US-Q1,
hybrid partitioning slightly outperforms kd-tree partitioning. Both hybrid
partitioning and kd-tree partitioning perform much better than metric-based
partitioning. It is because that most keywords in STS-US-Q1 are frequent among
objects, which results in a larger workload using text-partitioning. This can
also explain why hybrid partitioning and kd-tree partitioning have similar
performance. Because hybrid partitioning wisely choose space-partitioning to
partition the workload in most regions. In Figure 7(b), hybrid partitioning
and metric-based partitioning have a larger throughput than kd-tree
partitioning does. This can be explained in two aspects. First, the larger
query range results in a larger workload for kd-tree partitioning. Second, the
keywords in STS-US-Q2 and STS-UK-Q2 are less frequent, which improves the
performance of text-partitioning.

Latency The latency is the average time of each tuple staying in the system.
We evaluate all the algorithms using a moderate input speed of the data
stream.

(a) #Queries=5M (Q1)

(b) #Queries=10M (Q2)

(c) #Queries=10M (Q3)

Figure 8: Latency Comparison

Figure 8 shows the latencies of hybrid partitioning, metric-based partitioning
and kd-tree partitioning. Algorithm hybrid partitioning has smaller latency.
The latency of kd-tree partitioning is noticeably larger than hybrid
partitioning and metric-based partitioning in Figure 8(b), e.g., 25ms versus
15ms. This is caused by the larger spatial ranges of queries in STS-US-Q2 and
STS-UK-Q2. Algorithm metric-based partitioning has similar latency values as
hybrid partitioning does with the exception on STS-UK-Q1, which takes 407ms.
This is caused by the poor performance of text-partitioning when query
keywords are frequent. Figure 8 demonstrates that hybrid partitioning achieves
small latency for workload processing.

(a) #Queries=5M (Q1)

(b) #Queries=10M (Q2)

(c) #Queries=10M (Q3)

Figure 9: Memory Comparison (Dispatchers)

Memory Figure 9 shows the average memory usages of dispatchers of hybrid
partitioning, metric-based partitioning and kd-tree partitioning. Overall, the
memory usage of all the methods is not large, e.g., less than 1000MB, which is
acceptable in practice. Algorithm kd-tree partitioning uses less memory than
metric-based partitioning and hybrid partitioning. The memory usage of hybrid
partitioning is the highest for Q2 queries than for Q1 and Q3. This is because
that, for Q2 queries, more cells in the gridt index need to maintain
additional text partitioning information.

(a) #Queries=5M (Q1)

(b) #Queries=10M (Q2)

(c) #Queries=10M (Q3)

Figure 10: Memory Comparison (Workers)

Figure 10 shows the average memory usages of workers of the three methods.
Algorithm hybrid partitioning has an overall best performance, which invokes
the smallest memory usage in most cases. This is due to the reason that hybrid
partitioning distributes STS queries to workers considering the data
distributions in different regions, which reduces the cases of one STS query
being stored in multiple workers. We also observe that all the three methods
do not impose large memory requirements on workers.

(a) #Queries=10M (STS-UK-Q1)

(b) #Queries=20M (STS-UK-Q2)

(c) #Queries=20M (STS-UK-Q3)

Figure 11: Scalability

Scalability Figure 11 shows the throughputs of the evaluated algorithms as we
vary the number of workers. As we can see, hybrid partitioning exhibits the
best performance in most cases, and scales well with the number of workers.
Algorithm metric-based partitioning has the worst scalability in Figure 11(a)
as the keywords in STS-UK-Q1 are frequent, which results in a larger workload
on the workers. Figure 11(b) shows that kd-tree partitioning has the worst
scalability.

###  VI-D Evaluating Dynamic Load Adjustment

(a) Time of Selecting Cells

(b) Migration Cost and Time

(c) Percentage of Latency Values

Figure 12: Migration Experiments (#Q=1M, STS-US-Q1)

In this set of experiments, we evaluate the performance of our dynamic load
adjustment algorithms. The experiments are conducted on 4 dispatchers and 8
workers. In addition to our proposed dynamic programming (DP) algorithm and
greedy (GR) algorithm, we implement two other algorithms for comparison.

SI: It is another greedy algorithm which adds cells into $G_{s}$ (the set of
cells to be migrated), in descending order of their sizes.

RA: It selects the cells to be migrated randomly.

Comparing the running time of selecting cells for migration We run the four
algorithms on the same worker (having the same set of cells) and measure their
running time for selecting cells for migration, and the results are shown in
Figure 12(a) and Figure 13. Note that workers run out of memory when running
DP on queries used for the experiments in Figure 13. As shown in Figure 12(a),
DP runs significantly longer than the other algorithms. This is due to the
high time complexity of DP. As expected, RA is the fastest among all the
algorithms. However, both GR and SI are very efficient too. By comparing
Figures 13(a) and 13(b) , we can see that the running time of GR, SI and RA
does not change with the number of queries. The reason is that their running
time is only determined by the number of cells.

Comparing the migration cost We conduct the evaluation by running a workload
consisting of processing a sample of spatio-textual tweets in 60 days and the
insertions/deletions of STS queries. Since our tweets are only a small sample
of real-life tweets, we scale out the workload by reading 4 hours of tweets in
every 10 seconds. Note that we utilize the timestamps of tweets for the scale-
out. In each workload migration, we measure the size of migrated data and the
time in doing migration.

We report the average size of the migrated data and the average time in doing
migration in Figure 12(b) and Figure 15. In each figure, the left y-axis shows
migration cost and the right shows time. In Figure 12(b), DP and GR incur the
smallest migration cost and require the least time for migration. DP requires
slightly longer time than GR, since DP needs longer time in selecting the
cells to be migrated. GR performs the best as shown in Figure 15. In Figure
14(a), it incurs $30\%$–$40\%$ less migration cost than SI and RA. It takes
the least amount of time in doing migration, which is nearly half of the time
required by RA and $70\%$ of the time required by SI. Note that the migration
cost (resp. migration time) in Figure 14(b) is larger than the migration cost
(resp. migration time) in Figures 14(a). This is because that a larger number
of STS queries impose heavier load on the system and the size of each cell to
be migrated becomes larger too.

(a) #Queries=5M

(b) #Queries=10M

Figure 13: Average Time of Selecting Cells (STS-US-Q1)

(a) #Queries=5M

(b) #Queries=10M

Figure 14: Migration Cost and Time (STS-US-Q1)

(a) #Queries=5M

(b) #Queries=10M

Figure 15: Percentage of Latency Values (STS-US-Q1)

Comparing latency To further evaluate the effect of different algorithms on
the system, we also measure the processing latency. The results are presented
in Figure 12(c) and Figure 15. In Figure 12(c), GR has the smallest side
effect on the system, where $80\%$ of tuples are not affected by the migration
operations (less than 100ms) and $4\%$ of tuples are significantly delayed
(larger than 1 second). DP is second to GR, where $72\%$ of tuples are not
affected by the migration operations. It results in a larger fraction of
tuples being delayed between 100ms and 1 second. This is due to its long
running time in selecting cells to be migrated. SI and RA perform worse: SI
results in $10\%$ more tuples being delayed than GR. RA results in $20\%$ more
tuples being delayed than GR.

Evaluating the effect of dynamic load adjustments We evaluate the benefits of
our dynamic load adjustment algorithms by comparing the throughputs of the
system conducting dynamic load adjustments, and the system without conducting
dynamic load adjustments. We use our proposed GR algorithm in the dynamic load
adjustments. To simulate a workload with varying data distributions over time,
we use query set STS-US-Q3 for this experiment, and every interval of 10M
queries, the types of queries in 10% of the regions switch between STS-US-Q1
and STS-US-Q2. We set $\mu=10$M in this experiment.

Figure 16: The Effect of the Dynamic Load Adjustments

Figure 16 shows the experimental results. The system with dynamic load
adjustments outperforms the system without dynamic load adjustments by $26\%$
in terms of throughput. The results demonstrate the effectiveness of our
dynamic load adjustment algorithms.

##  VII Conclusion

In this work, we develop an efficient distributed publish/subscribe system,
PS2Stream, over a spatio-textual data stream. We consider the optimal workload
partitioning problem and propose a new hybrid partitioning algorithm. We also
propose effective dynamic load adjustment approaches to adjust the load of
workers in the scenario of workload changing. Our experimental results show
that our workload distribution framework performs better than the baselines in
both throughput and latency, and our dynamic load adjustment approaches
improve the performance of the system with small migration cost.

## Acknowledgement

This work was carried out at the Rapid-Rich Object Search(ROSE) Lab at the
Nanyang Technological University, Singapore. The ROSE Lab is supported by the
National Research Foundation, Singapore, under its Interactive Digital
Media(IDM) Strategic Research Programme. This work is also supported in part
by a Tier-1 grant (RG22/15) and a Tier-2 grant (MOE-2016-T2-1-137) awarded by
Ministry of Education Singapore, and a grant awarded by Microsoft. Tom Fu and
Zhenjie Zhang are supported by the research grant for the Human-Centered
Cyber-physical Systems Programme at the Advanced Digital Sciences Center from
Singapore’s Agency for Science, Technology and Research (A*STAR). They are
also partially supported by Science and Technology Planning Project of
Guangdong under grant (No. 2015B010131015). We also acknowledge the “AWS Cloud
Credits for Research” program for its support of this work.

## References

  * [1] F. Cao and J. P. Singh, “Efficient event routing in content-based publish-subscribe service networks,” in _INFOCOM_. IEEE, 2004, pp. 929–940.
  * [2] M. A. Tariq, B. Koldehofe, G. G. Koch, and K. Rothermel, “Providing probabilistic latency bounds for dynamic publish/subscribe systems,” in _Kommunikation in Verteilten Systemen (KiVS)_. Springer, 2009, pp. 155–166.
  * [3] O. Papaemmanouil, Y. Ahmad, U. Çetintemel, J. Jannotti, and Y. Yildirim, “Extensible optimization in overlay dissemination trees,” in _SIGMOD_ , 2006, pp. 611–622.
  * [4] R. Baldoni, C. Marchetti, A. Virgillito, and R. Vitenberg, “Content-based publish-subscribe over structured overlay networks,” in _ICDCS_. IEEE, 2005, pp. 437–446.
  * [5] A. Carzaniga, D. S. Rosenblum, and A. L. Wolf, “Design and evaluation of a wide-area event notification service,” _TOCS_ , vol. 19, no. 3, pp. 332–383, 2001.
  * [6] A. Yu, P. K. Agarwal, and J. Yang, “Subscriber assignment for wide-area content-based publish/subscribe,” _TKDE_ , vol. 24, no. 10, pp. 1833–1847, 2012.
  * [7] G. Li, Y. Wang, T. Wang, and J. Feng, “Location-aware publish/subscribe,” in _KDD_. ACM, 2013, pp. 802–810.
  * [8] H. Hu, Y. Liu, G. Li, J. Feng, and K. Tan, “A location-aware publish/subscribe framework for parameterized spatio-textual subscriptions,” in _ICDE_ , 2015, pp. 711–722.
  * [9] X. Wang, Y. Zhang, W. Zhang, X. Lin, and W. Wang, “Ap-tree: Efficiently support continuous spatial-keyword queries over stream,” in _ICDE_ , pp. 1107–1118.
  * [10] L. Chen, G. Cong, and X. Cao, “An efficient query indexing mechanism for filtering geo-textual data,” in _SIGMOD_. ACM, 2013, pp. 749–760.
  * [11] X. Wang, Y. Zhang, W. Zhang, X. Lin, and Z. Huang, “Skype: top-k spatial-keyword publish/subscribe over sliding window,” _Proceedings of the VLDB Endowment_ , vol. 9, no. 7, pp. 588–599, 2016.
  * [12] M. Yu, G. Li, and J. Feng, “A cost-based method for location-aware publish/subscribe services,” in _Proceedings of the 24th ACM International on Conference on Information and Knowledge Management_. ACM, 2015, pp. 693–702.
  * [13] L. Chen, G. Cong, X. Cao, and K. Tan, “Temporal spatial-keyword top-k publish/subscribe,” in _ICDE_ , 2015, pp. 255–266.
  * [14] W. W. Terpstra, S. Behnel, L. Fiege, A. Zeidler, and A. P. Buchmann, “A peer-to-peer approach to content-based publish/subscribe,” in _DEBS_ , 2003\.
  * [15] A. Gupta, O. D. Sahin, D. Agrawal, and A. El Abbadi, “Meghdoot: Content-based publish/subscribe over P2P networks,” in _Middleware_ , 2004, pp. 254–273.
  * [16] R. Barazzutti, T. Heinze, A. Martin, E. Onica, P. Felber, C. Fetzer, Z. Jerzak, M. Pasin, and E. Rivière, “Elastic scaling of a high-throughput content-based publish/subscribe engine,” in _Distributed Computing Systems (ICDCS), 2014 IEEE 34th International Conference on_. IEEE, 2014, pp. 567–576.
  * [17] A. Aji, F. Wang, H. Vo, R. Lee, Q. Liu, X. Zhang, and J. Saltz, “Hadoop gis: a high performance spatial data warehousing system over mapreduce,” _Proceedings of the VLDB Endowment_ , vol. 6, no. 11, pp. 1009–1020, 2013\.
  * [18] A. Eldawy and M. F. Mokbel, “Spatialhadoop: A mapreduce framework for spatial data,” in _2015 IEEE 31st International Conference on Data Engineering_. IEEE, 2015, pp. 1352–1363.
  * [19] A. Akdogan, U. Demiryurek, F. Banaei-Kashani, and C. Shahabi, “Voronoi-based geospatial query processing with mapreduce,” in _Cloud Computing Technology and Science (CloudCom), 2010 IEEE Second International Conference on_. IEEE, 2010, pp. 9–16.
  * [20] S. Nishimura, S. Das, D. Agrawal, and A. E. Abbadi, “Md-hbase: a scalable multi-dimensional data infrastructure for location aware services,” in _MDM_ , vol. 1. IEEE, 2011, pp. 7–16.
  * [21] A. M. Aly, A. R. Mahmood, M. S. Hassan, W. G. Aref, M. Ouzzani, H. Elmeleegy, and T. Qadah, “Aqwa: adaptive query workload aware partitioning of big spatial data,” _Proceedings of the VLDB Endowment_ , vol. 8, no. 13, pp. 2062–2073, 2015.
  * [22] J. Chen, D. J. DeWitt, F. Tian, and Y. Wang, “Niagaracq: A scalable continuous query system for internet databases,” in _ACM SIGMOD Record_ , vol. 29, no. 2. ACM, 2000, pp. 379–390.
  * [23] S. Chandrasekaran and M. J. Franklin, “Psoup: a system for streaming queries over streaming data,” _The VLDB Journal_ , vol. 12, no. 2, pp. 140–156, 2003\.
  * [24] M. Cherniack, H. Balakrishnan, M. Balazinska, D. Carney, U. Cetintemel, Y. Xing, and S. B. Zdonik, “Scalable distributed stream processing.” in _CIDR_ , vol. 3, 2003, pp. 257–268.
  * [25] D. J. Abadi, Y. Ahmad, M. Balazinska, U. Cetintemel, M. Cherniack, J.-H. Hwang, W. Lindner, A. Maskey, A. Rasin, E. Ryvkina _et al._ , “The design of the borealis stream processing engine.” in _CIDR_ , 2005, pp. 277–289.
  * [26] A. R. Mahmood, A. M. Aly, T. Qadah, E. K. Rezig, A. Daghistani, A. Madkour, A. S. Abdelhamid, M. S. Hassan, W. G. Aref, and S. Basalamah, “Tornado: A distributed spatio-textual stream processing system,” _PVLDB_ , vol. 8, no. 12, pp. 2020–2023, 2015.
  * [27] B. B. Cambazoglu, E. Kayaaslan, S. Jonassen, and C. Aykanat, “A term-based inverted index partitioning model for efficient distributed query processing,” _TWEB_ , vol. 7, no. 3, p. 15, 2013.
  * [28] F. Basık, B. Gedik, H. Ferhatosmanoğlu, and M. E. Kalender, “S3-tm: scalable streaming short text matching,” _The VLDB Journal_ , vol. 24, no. 6, pp. 849–866, 2015.
  * [29] S. Vaid, C. B. Jones, H. Joho, and M. Sanderson, “Spatio-textual indexing for geographical search on the web,” in _SSTD_ , 2005, pp. 218–235.
  * [30] V. Gulisano, R. Jimenez-Peris, M. Patino-Martinez, and P. Valduriez, “Streamcloud: A large scale data streaming system,” in _ICDCS_. IEEE, 2010, pp. 126–137.


# COVID Detection in Chest CTs: Improving the Baseline on COV19-CT-DB

Radu Miron
Cosmin Moisii
Sergiu Dinu
Mihaela Breaban (✉)  SenticLab, Iasi, Romania
Faculty of Computer Science, ”Alexandru Ioan Cuza” University of Iasi, Romania
<EMAIL_ADDRESS>
###### Abstract

The paper presents a comparative analysis of three distinct approaches based
on deep learning for COVID-19 detection in chest CTs. The first approach is a
volumetric one, involving 3D convolutions, while the other two approaches
perform at first slice-wise classification and then aggregate the results at
the volume level. The experiments are carried on the COV19-CT-DB dataset, with
the aim of addressing the challenge raised by the MIA-COV19D Competition
within ICCV 2021. Our best results on the validation subset reach a macro-F1
score of 0.92, which improves considerably the baseline score of 0.70 set by
the organizers.

##  1 Introduction

There is a high effervescence in the AI community trying to provide enhanced
tools to assist the medical diagnosis, with a special focus on medical
imaging, exploiting but also triggering important advancements in the area of
computer vision and deep learning. The pandemic scenario we face today clearly
calls for such approaches in order to be able to address the high incidence
rate which overwhelms the medical system. Medical imaging, mostly in the form
of x-Rays and CTs, is used to assess the lung involvement. In this context,
COVID-19 datasets are publicly released and competitions are organized, aiming
at involving and stimulating the AI community to produce models that can
accurately detect Covid-19 affections in the lungs.

The current work addresses the challenge raised by the MIA-COV19D
competition111https://mlearn.lincoln.ac.uk/mia-cov19d/ organized within the
ICCV 2021 conference. The challenge consists in tackling a two-class
classification problem on the COV19-CT-DB dataset consisting of CTs classified
in two groups: COVID patients and non-COVID patients, with the second class
containing both healthy patients or patients presenting lung lesions due to
other causes. The dataset was split by the organizers into 3 subsets:
training, validation and test, with the first two subsets also exposing the
class label. The baseline set for this classification task by the organizers,
for comparison purposes, is a 0.70 macro-F1 score on the validation set,
obtained with a CNN-RNN network described in [5], which is the result of
authors’ previous work reported in [7, 6, 8]. The final evaluation of the
models built by the participants in the competition will be made on the test
set that was released with no class/label information.

Given the nature of the data, we address the challenge in two different ways:
1) treating the CT as a volume and thus using 3D convolutions, and 2) treating
the CT as a set of 2d images (slices), the second approach calling for 2D
convolutions and predictions at slice level, followed by an aggregation step
where the outputs at slice level are aggregated into a response at CT level.

Our second approach is motivated by the results we obtained in the 2020 and
2021 ImageClefMed competitions on tuberculosis tasks, where our team was
ranked the first [10, 11]. This approach needs in the training phase more
refined information: it needs CT slices split in two categories: slices that
present COVID lesions and slices that do not present COVID lesions (meaning
healthy or other affection types). Because the dataset exposes this
information only at CT level and a COVID case presents both healthy and
affected lung slices, we needed to come up with a procedure to extract COVID-
affected slices from the COVID CTs in the training set. We use at first a
naive method to label slices, where all the slices belonging to a COVID CT are
considered as slices presenting COVID lesions. As a second more refined
method, we use a model we had trained for the tuberculosis classification task
in the ImageClefMed 2021 competition that is able to detect slices presenting
different types of tuberculosis lesions. Since the ImageClef data does not
contain any COVID patients, we expect that the results are not highly accurate
in identifying slices presenting COVID affections and this approach also
introduces noise, mainly as slices presenting other affections and not COVID,
because these could be present in both COVID and non-COVID CTs. After slice
labeling, we use two distinct approaches for COVID classification: one that
makes use of a manually designed NN architecture working with ”mini-volumes”
made of 3 slices and one that performs NN architecture search with
sharpDARTS[4].

The paper is structured as follows. Section 2 describes the COV19-CT-DB
dataset. Section3 describes our first approach that makes inference directly
at the CT level. Sections 4 and 5 describe the two approaches that learn to
classify the slices as COVID/non-COVID and then aggregate the results at CT
level. Section 6 presents the results obtained on the training and validation
sets and section 7 concludes the paper.

##  2 The dataset

The only dataset used in the experimental analysis is COV19-CT-DB [5],
provided in the MIA-COV19D competition.

Our slice-based approaches described in sections 4 and 5 do not involve
directly an external dataset, but only a pretrained model built by us in the
ImageClef2021 competition on CT data consisting of non-COVID patients that
present tuberculosis lesions; this pretrained model serves only to select a
subset of slices from the COVID class in the COV19-CT-DB training dataset. We
emphasise the fact that using this pretrained model built actually for a
tuberculosis task and used here to select slices presenting COVID might
introduce noise in the training set we build at slice level.

The provided COV19-CT-DB training set has a total of 1560 CT scans. The class
distribution is 690 COVID-19 cases versus 870 Non-COVID19 cases. Out of these,
we found 8 volumes with less than 20 slices and 5 volumes with more than 700
slices in their corresponding directories.

The provided validation set has a total of 374 CT scans. The class
distribution is 165 COVID-19 cases versus 209 Non-COVID-19 cases. Out of
these, we found no volume with less than 20 slices and 1 volume with more than
700 slices in their corresponding directories.

As explained in [5], the labeling of the dataset was made by a consensus of
two radiologists and two pneumologists. The difference with other publicly
available datasets is the annotations based on expert opinion rather than just
positive RT-PCR testing.

The whole COV19-CT-DB database consists of about 5000 CT scans, corresponding
to more than 1000 patients and 2000 subjects, leading to 3455 cases for the
test set.

The images are provided through jpg format by clipping the voxel intensities
using a window/level of 350 Hounsﬁeld units (HU) and 1150 HU and normalization
to the range of [0,1] of each slice from a whole volume rather than the dicom
files.

Due to the large number of slices per volume and large dimensions of an
individual slice, in our approaches we usually rescale the given images. Each
classification approach described in the sections below will also list the
modifications done to the input data.

##  3 A volumetric approach

Convolutional neural networks have emerged as a successful tool in tackling a
wide range of image vision tasks. These architectures have obtained, ever
since the appearance of Alex-Net [9], state of the art results on tasks like
image classification, object detection and segmentation.

Even though applications in 2D image tasks have shown great results,
architectures for 3D image tasks still have to prove themselves as good
contenders for state of the art results.

Nowadays benchmarks use pretraining for improving results. Since large 3D
datasets are not so numerous, we turn our attention to a specific kind of
model to fully benefit of the advantages of pretrained models.

We use an inflated convolutional neuronal network pretrained on Kinetics
dataset [1]. Inflated convolutions are obtained by expanding filters and
pooling kernels of 2D ConvNets into 3D, resulting in the possibility to learn
spatio-temporal feature extractors from 3D images while using successful
ImageNet architectures. Due to gpu limitations we only used an Inflated
ResNet50 model [3].

For capturing long range dependencies within the slices of a same volume, we
use non-local features which have been proven to increase the results of basic
architectures. [14]

The final model is an Inflated 3D ResNet50 with non local operations on the
second and third layers (based on the official implementation from PyTorch).
222https://pytorch.org/vision/0.8/_modules/torchvision/models/resnet.html

Due to large number of parameters, 3D conv nets are easy to overfit. In order
to mitigate the overfitting, we use label smoothing [12] with cross entropy
and Sharpness Aware Minimization [2] on top of SGD as loss function and
optimizer, respectively. We also use augmentations like Random Horizontal and
Vertical Flip, Gaussian Blur and Contrast and Color changes. We also tried
with Cut-Out and Affine transformations, but the results were not improved
compared to basic augmentations mentioned above. We may also flip the volume
on the depth axis. All these augmentations are done with 0.5 chance.

The training procedure lasts for 150 epochs. Initial learning rate is set to
1e-3 and the learning rate scheduler is cosine scheduler with 5 epochs of
warm-up. Batch size is 2.

The input consists of volumes of size $128\times 224\times 224$. First, we
resize all the images to $224\times 224$ and take only $128$ slices from each
volume (with padding if necessary). Since there are large volumes in our
dataset, we solve this issue using the following rule: if a volume has between
$128k$ and $128(k+1)$ slices, we choose the starting slice a random number
between $0$ and $k$ and sample every $k^{th}$ slice, discarding the others,
generating a volume of 128 slices. This way, we make sure that most of the
volume is preserved. Our procedure i based on the fact that volumes with large
number of slices (corresponding to small slice thickness) admit a transversing
with a small sliding window without great loss of information for covid
diagnosis.

In order to support the results we obtained with the validation dataset
provided by the organizers, we also created another 4 folds and completed for
each fold the training and validation steps.

During the inference process, parts of a single volume will be several times
input for the model. If a volume has between $256k$ and $256(k+1)$ slices, we
choose as starting slice each number between $0$ and $k$ and take each
$k^{th}$ slice discarding the others. This procedure generates $k+1$ sub-
volumes of 256 slices at inference time. For each sub-volume we apply
horizontal or vertical flip or depth axis flip. This means $8(k+1)$ inferences
for a certain volume. For inference we define two important thresholds: the
confidence threshold for non-covid label and confidence threshold for both
labels. Every non-covid prediction which is below the first threshold is
ignored. The same happens for the second threshold but with both of the
labels. After the eliminations based on the thresholding, the most frequent
diagnosis is given to the specific volume.

For the final prediction pipeline, we constructed an ensemble with 2 models
from each fold that we trained on, stored at the epochs we obtained the
highest validation accuracy on.

Some of the results of folds $2-5$ are mentioned in the table below:

Fold/Epoch | F1 score
---|---
$2/93$ | $0.936$
$3/99$ | $0.923$
$4/99$ | $0.936$
$5/96$ | $0.925$
official / ensemble | $0.9234$

##  4 A slice based approach with mini-volumes

This method is a slice based approach with the addition that it incorporates
the immediate previous and next slice creating a ”minivol”. It uses both slice
and lung (side) level labels which we obtained in two steps:

  * •

a neural network trained on an external dataset but with slice level labels
(the dataset in question is a small subset of the training set for the
ImageCLEF 2021 - Tuberculosis detection competition [11], which we manually
labeled)

  * •

manual correction of the predictions and further labeling

As the COV19-CT-DB dataset it too big to manually address slice labeling, we
applied this process to only a small subset of the data. The final dataset we
construct will have 3 labels: COVID, non-COVID pneumonia, healthy.

To use the aforementioned neural network trained at ImageClef, we need to
apply on the COV19-CT-DB data the same pre-processings used for its training.
This turned out to be a complex and imprecise task as the original ImageClef
training set used CT volumes in DICOM format and some of the pre-processing
steps were applied on the Hounsfield values of the input, while the COV19-CT-
DB is in jpg format.

Next, we shortly describe the process involved in building the model for
tuberculosis classification in the ImageClef competition, in order to
understand its use for building the COVID refined dataset here. Given a
selected slice in a CT, we grouped it together with the previous and the next
slice in the volume; these mini volumes of 3 consecutive slices, we thought,
could better highlight the lesions present in the tuberculosis dataset,
emphasising the difference between an infiltration and an artery, or a cavern
and a lumen as these can be very similar at a certain point in space but
continue in a different manner. We changed its window and level values to
highlight the lung features. The selected slices were split into half,
corresponding to each lung, and we kept only the side that was labeled. We
cropped the images, using a simple threshold method to remove the padding and
keep only the body. The resulting images were resized to $256\times 256$
pixels. These were then normalized with values in the range $[0,255]$
corresponding to 3 black and white images which were concatenated at channel
level. As augmentations we used a random crop of size $224\times 224$, a
random horizontal flip with a probability of $0.5$ and normalized the image.
We trained an EfficientNet-B3 [13], with batches of 32 for 90 epochs.

We used this network to predict on each slice of a volume for the COVID
patients in the COV19-CT-DB training set the presence of an affection, and
obtain the probabilities of each affection type that was present in the
ImageClef data (a total of five lesion types and one healthy class).

In order to be able to apply this pre-trained model on the COVID dataset we
had to address several problems. The steps we followed to address the pre-
processing are:

  * •

We aggregated the slices into volumes

  * •

Constructed an equivalent cropping functions based on pixel thresholds

  * •

Made a prediction with the pre-trained neural network and filtered the results
based on one specific label - ”healthy”. Using an upper and a lower threshold
we filtered out the predictions with a score for the ”healthy” label between
those values, thus keeping only high confidence predictions. We also filtered
based on slice location, preferring slices at the middle at the series so as
not to saturate the ”healthy” class with slices at the ends of the volume that
were by default healthy as they did not contain yet the lungs.

  * •

We retrained and made the predictions on the entire original training set. In
order to obtain the same number of predictions per patients for the last
training we resized the reconstructed volumetric image to a depth of size
$96$. As some of the patients’ folder have very few slices this means the
resize would just multiply the same images along the volume.

  * •

For each slice we predicted the probabilities of the 3 classes enumerated
above. We grouped the slices predictions into an array of size $(96,3)$. We
did this for each patient, thus creating a data set that was used as training
data for a linear classifier that learns to predict the class at CT level.

To predict on the validation set we would first have to predict with the first
neural network on the reconstructed slices at slice level and then predict on
the results with the second classifier at patient level. We trained a logistic
regression classifier and a multilayer perceptron with 100 neurons and a
single layer. The results on the training and validation sets are shown in
tables 2 and 3.

##  5 A slice based approach tackled with neural network architecture search

This approach is intended to act as our own baseline, as it consists of very
simple steps that could be taken to build a classifier for CT data:

  * •

label each CT slice in the training set with its corresponding CT label;

  * •

use an existing implementation that performs neural architecture search for
image classification to generate a network architecture that performs well on
our problem at hand - classifying the CT slices;

  * •

aggregate the results at slice level using simple statistics in order to
classify a CT.

In the last few years there have been great improvements in the field of
Neural Architecture Search (NAS). The state-of-the-art models are able to find
neural network architectures in a small number of GPU days, much less than the
long search times that were required just a few years ago. Since we are
dealing with a classification problem, employing a neural architecture search
algorithm that has a very good result on a benchmark classification dataset is
an idea worth exploring.

The current state-of-the-art neural architecture search algorithm for the
classification of the cifar-10 dataset is sharpDARTS, introduced in [4]. The
sharpDARTS algorithm operates in such a way that it searches for a primitive
that will appear multiple times in a larger neural network architecture.

The authors of the sharpDARTS algorithm are modelling the space of the neural
network architectures in such a way that is it differentiable. In this
differentiable search space, gradient based search methods are proven to be
very effective. The authors have also made the implementation of sharpDARTS
available 333https://github.com/ahundt/sharpDARTS. Starting from the authors’
implementation, we have made the necessary adjustments to the implementation
in order to run the search on the present dataset. In this approach, the
classification takes place at the level of each CT slice. We have downsampled
the resolution of the images from $512$x$512$ to $192$x$192$ in order to speed
up the training process by avoiding the loading in memory of the entire
dataset multiple times.

We build the slice-based dataset in a very simple manner, being aware that it
has some important flaws: every label in a COVID CT is marked as COVID, while
every label in a non-COVID CT is marked as non-COVID. No filtering is
performed, meaning that we expect to have many similar slices in the dataset
with contradictory labels, coming from both COVID patients and healthy
patients, since not all the slices in a COVID CT present COVID specific
lesions.

SharpDARTS’ authors have found a state-of-the-art model on the cifar-10
dataset in 0.8 days. We have run the search algorithm for 1 day, finding a
primitive presented in Figure 1.

Figure 1: The primitive found by the sharpDARTS search algorithm.

With this primitive, we have constructed the final architecture as shown in
Figure 2.

Figure 2: The final architecture found by the sharpDARTS search algorithm.

We continued to train the final neural network architecture for $300$ epochs.

We present the results of this per-slice approach in Table 1 \- scores are
computed here based on slice labels and not CT labels (we measure the
performance of the model for labeling slices).

We compute the label for each volume by utilizing the majority vote rule,
based on the labels predicted for all the slices in the volume. The results
for CTs classification are presented in Tables 2 and 3. Despite it being a
relatively small architecture of only $61.3$ MMAC (million multiply-accumulate
operations; $122.6$ MFLOPS) and being trained on noisy data, it obtained an
unexpectedly good F1 score on the validation dataset of 0.74.

Data | precision | recall | macro F1
---|---|---|---
Train | $0.724$ | $0.731$ | $0.727$
Validation | $0.729$ | $0.777$ | $0.752$
Table 1: Per slice results of the sharpDARTS algorithm

It is worth improving this approach by utilizing the slice labeling procedure
described in Section 4. This would help eliminate the issue of slices that are
not informative relative to the covid or non-covid nature of the CT. We would
also like to remind that this approach is at a disadvantage since it only
utilizes a single slice at a time, losing context information.

##  6 Experimental results

Tables 2 and 3 present the results obtained on the training and validation
sets, in terms of precision and recall for the COVID class, and the macro F1
score, as required in the competition.

All the approaches taken outperform the baseline score set by the competition
organizers [5].

The volumetric approach achieves the best results, bringing a consistent
improvement over the baseline.

Although with lower accuracy, we consider the slice-based approaches to be
more relevant in practice, providing more information related to lesion
localization in the lungs and implicitly its extension/size. Given our
previous experience on similar CT classification tasks [10, 11], we argue that
the poorer results are due to the presence of noisy labels at slice level in
the training data (because of the automated slice labeling procedures we use)
and filtering manually COVID slices is necessary to build a good training set
for increasing the performance of this approach.

Method | precision (Covid class) | recall (Covid class) | macro F1 score
---|---|---|---
volumetric approach | $0.97$ | $0.96$ | $0.96$
slice-wise approach based on minivolumes & log reg | $0.95$ | $0.89$ | $0.93$
slice-wise approach based on minivolumes & mlp | $0.98$ | $0.93$ | $0.97$
slice-wise approach based on sharpDARTS | $0.93$ | $0.94$ | $0.93$
Table 2: Results on the train set for: the volumetric approach described in Section 3, the two slice based approaches based on mini-volumes described in Section 4 and for the slice based approach using neural architecture search described in Section 5 Method | precision (Covid class) | recall (Covid class) | macro F1 score
---|---|---|---
volumetric approach | $0.95$ | $0.88$ | $0.92$
slice-wise approach based on minivolumes & log reg | $0.83$ | $0.71$ | $0.82$
slice-wise approach based on minivolumes & mlp | $0.87$ | $0.76$ | $0.84$
slice-wise approach based on sharpDARTS | $0.67$ | $0.82$ | $0.74$
Table 3: Results on the validation set for: the volumetric approach described
in Section 3, the two slice based approaches based on mini-volumes described
in Section 4 and for the slice based approach using neural architecture search
described in Section 5

##  7 Conclusions

The paper presents the approaches taken by the SenticLab.UAIC team for the
problem of COVID19 detection in CTs. Given the nature of the data, consisting
of volumetric (3D) images, two distinct ways to perform CT classification are
used: one that treats the CT as a whole, and one that performs classification
at slice level and then aggregates the results at CT level. The experiments
show very promising results, reaching a 0.92 macro F1 score.

## References

  * [1] João Carreira and Andrew Zisserman.  Quo vadis, action recognition? A new model and the kinetics dataset.  CoRR, abs/1705.07750, 2017.
  * [2] Pierre Foret, Ariel Kleiner, Hossein Mobahi, and Behnam Neyshabur.  Sharpness-aware minimization for efficiently improving generalization.  CoRR, abs/2010.01412, 2020.
  * [3] Kaiming He, Xiangyu Zhang, Shaoqing Ren, and Jian Sun.  Deep residual learning for image recognition.  CoRR, abs/1512.03385, 2015.
  * [4] Andrew Hundt, Varun Jain, and Gregory D. Hager.  sharpdarts: Faster and more accurate differentiable architecture search.  CoRR, abs/1903.09900, 2019.
  * [5] Dimitrios Kollias, Anastasios Arsenos, Levon Soukissian, and Stefanos Kollias.  Mia-cov19d: Covid-19 detection through 3-d chest ct image analysis.  arXiv preprint arXiv:2106.07524, 2021.
  * [6] Dimitrios Kollias, N Bouas, Y Vlaxos, V Brillakis, M Seferis, Ilianna Kollia, Levon Sukissian, James Wingate, and S Kollias.  Deep transparent prediction through latent representation analysis.  arXiv preprint arXiv:2009.07044, 2020.
  * [7] Dimitrios Kollias, Athanasios Tagaris, Andreas Stafylopatis, Stefanos Kollias, and Georgios Tagaris.  Deep neural architectures for prediction in healthcare.  Complex & Intelligent Systems, 4(2):119–131, 2018.
  * [8] Dimitris Kollias, Y Vlaxos, M Seferis, Ilianna Kollia, Levon Sukissian, James Wingate, and Stefanos D Kollias.  Transparent adaptation in deep medical image diagnosis.  In TAILOR, pages 251–267, 2020.
  * [9] Alex Krizhevsky, Ilya Sutskever, and Geoffrey E Hinton.  Imagenet classification with deep convolutional neural networks.  In Advances in neural information processing systems, pages 1097–1105, 2012.
  * [10] Radu Miron, Cosmin Moisii, and Mihaela Breaban.  Revealing lung affections from cts. A comparative analysis of various deep learning approaches for dealing with volumetric data.  In Linda Cappellato, Carsten Eickhoff, Nicola Ferro, and Aurélie Névéol, editors, Working Notes of CLEF 2020 \- Conference and Labs of the Evaluation Forum, Thessaloniki, Greece, September 22-25, 2020, volume 2696 of CEUR Workshop Proceedings. CEUR-WS.org, 2020.
  * [11] Cosmin Moisii, Radu Miron, and Mihaela Breaban.  Identifying tuberculosis type in cts.  In Guglielmo Faggioli, Nicola Ferro, Alexis Joly, Maria Maistro, and Florina Piroi, editors, Working Notes of CLEF 2021 - Conference and Labs of the Evaluation Forum, 2021, CEUR Workshop Proceedings. CEUR-WS.org, 2021.
  * [12] Rafael Müller, Simon Kornblith, and Geoffrey E. Hinton.  When does label smoothing help?  CoRR, abs/1906.02629, 2019.
  * [13] Mingxing Tan and Quoc Le.  Efficientnet: Rethinking model scaling for convolutional neural networks.  In International Conference on Machine Learning, pages 6105–6114. PMLR, 2019.
  * [14] Xiaolong Wang, Ross Girshick, Abhinav Gupta, and Kaiming He.  Non-local neural networks, 2018.