Importing COVID-19 data into Elasticsearch

UPDATE

Due to a change in the way that John Hopkins Hospital organizes the COVID-19 related data the pipeline together with the index template have been changed. The below configurations for both the pipeline and template have been updated to work properly when importing COVID-19 data into elasticsearch. These changes should not impact the dashboard.

We launched an interactive Kibana dashboard for the COVID-19 outbreak. You can use the dashboard for filtering and comparing COVID-19 data from different countries, regions and counties enabling you to run your own analysis of the outbreak.

SEE DASHBOARD

The outbreak of COVID-19 virus has had a huge impact on both personal lives as well as business activities and in light of this, access to data is essential in allowing everyone to make informed decisions regarding day-to-day activities. John Hopkins Hospital, which is deeply involved in the international response to the virus, is collecting and publishing the data regarding the outbreak through GitHub as well as daily reports and a live dashboard.

In this blog post we will provide you with the details needed to create a logstash pipeline that brings the data collected by John Hopkins Hospital into your own Elasticsearch cluster. We also added the details for this process in a GitHub repository that can be found here.

If you are interested to see how Siscale can help you during these uncertain times check out this page. For more information on ingesting COVID-19 data read on!

Logstash

Below is the Logstash code required to collect, transform the data into time series data and import it into Elasticsearch. The code contains commented lines which explain what each part of the code does.

-- CODE markdown--
input {

#Use the generator input to update the dataset when the pipeline is first started
generator {
lines => [ "first-time-run" ]
count => 1
tags => "check_github"
}

#The http_poller input plugin is used to schedule checks for dataset updates
#The "schedule" setting is defined to check for updates every new hour (at minute 0)
http_poller {
urls => {
check_github => "https://github.com/CSSEGISandData/COVID-19/tree/master/csse_covid_19_data/csse_covid_19_daily_reports"
}
tags => "check_github"
request_timeout => 60
schedule => { "cron" => "0 * * * * UTC" }
codec => "plain"
metadata_target => "http_poller_metadata"
}
}

filter {

#The pipeline will treat two types of events:
#The first type is the initial event that triggers the downloading, parsing, and transforming of the CSVs into time series data
#The second type is the time series data itself
#This 'if' discriminates between the two types. The time series data is treated later
if "check_github" in [tags] {
ruby {
init => '
require "csv"
require "open-uri"
require "digest"
require "json"

#Path to store the MD5 hashes of the downloaded CSV files
#This is used to keep track of any changes to the Github repo and optimize the update process
@github_stored_hashes_path = "/etc/logstash/covid-19-hashes.json"

#Base URL for downloading the time series CSVs
@github_daily_reports_base_url = "https://raw.githubusercontent.com/CSSEGISandData/COVID-19/master/csse_covid_19_data/csse_covid_19_daily_reports/"

@states = { "AL" => "Alabama", "AK" => "Alaska", "AS" => "American Samoa", "AZ" => "Arizona", "AR" => "Arkansas", "CA" => "California", "CO" => "Colorado", "CT" => "Connecticut", "DE" => "Delaware", "DC" => "District of Columbia", "FM" => "Federated States of Micronesia", "FL" => "Florida", "GA" => "Georgia", "GU" => "Guam", "HI" => "Hawaii", "ID" => "Idaho", "IL" => "Illinois", "IN" => "Indiana", "IA" => "Iowa", "KS" => "Kansas", "KY" => "Kentucky", "LA" => "Louisiana", "ME" => "Maine", "MH" => "Marshall Islands", "MD" => "Maryland", "MA" => "Massachusetts", "MI" => "Michigan", "MN" => "Minnesota", "MS" => "Mississippi", "MO" => "Missouri", "MT" => "Montana", "NE" => "Nebraska", "NV" => "Nevada", "NH" => "New Hampshire", "NJ" => "New Jersey", "NM" => "New Mexico", "NY" => "New York", "NC" => "North Carolina", "ND" => "North Dakota", "MP" => "Northern Mariana Islands", "OH" => "Ohio", "OK" => "Oklahoma", "OR" => "Oregon", "PW" => "Palau", "PA" => "Pennsylvania", "PR" => "Puerto Rico", "RI" => "Rhode Island", "SC" => "South Carolina", "SD" => "South Dakota", "TN" => "Tennessee", "TX" => "Texas", "UT" => "Utah", "VT" => "Vermont", "VI" => "Virgin Islands", "VA" => "Virginia", "WA" => "Washington", "WV" => "West Virginia", "WI" => "Wisconsin", "WY" => "Wyoming" }

#Input: a CSV and the date of the CSV
#Output: an array of Logstash events extracted from the CSV that will be sent to Elasticsearch
def create_events(csv, current_date)
totals_by_country = Hash.new
totals_by_country["WORLD"] = LogStash::Event.new
totals_by_country["WORLD"].set("total_confirmed", 0)
totals_by_country["WORLD"].set("total_recovered", 0)
totals_by_country["WORLD"].set("total_deaths", 0)
totals_by_country["WORLD"].tag("total_world")
totals_by_country["WORLD"].tag("covid_time_series")
totals_by_country["WORLD"].set("timestamp", current_date.iso8601(3).to_s)
totals_by_country["WORLD"].set("unique_id", "totals_for_world_" + current_date.strftime("%m-%d-%Y").to_s.gsub(/[^\w\d]/, "_"))
all_country_region = Array.new
all_province_state = Array.new
all_county = Array.new
all_country_region.append("WORLD")

events_array = Array.new
#For each event..
csv.each do |row|
#..create a new Logstash event that will store the data
new_event = LogStash::Event.new

#..add current row data to the current event
row.each do |k, v|
#new_event[k] = v
new_event.set(k, v&.strip)
end

#..store the date of the reporting
new_event.set("timestamp", current_date.iso8601(3).to_s)
#..store the date when Logstash processed the event
new_event.set("last_process", Time.now.iso8601(3).to_s)

#..some early events contain empty fields when the reported numbers are 0. Fix this here
new_event.set("confirmed", 0) if new_event.get("confirmed").nil?
new_event.set("recovered", 0) if new_event.get("recovered").nil?
new_event.set("deaths", 0) if new_event.get("deaths").nil?
#..calculate the number of active cases, if the field is not already populated
active = new_event.get("confirmed").to_i - new_event.get("recovered").to_i - new_event.get("deaths").to_i
new_event.set("active", active) if active >= 0

#..calculate ratios and store them
ratio_deaths_to_confirmed = new_event.get("deaths").to_f / new_event.get("confirmed").to_f
ratio_recovered_to_confirmed = new_event.get("recovered").to_f / new_event.get("confirmed").to_f
ratio_deaths_to_recovered = new_event.get("deaths").to_f / new_event.get("recovered").to_f
new_event.set("ratio_deaths_to_confirmed", ratio_deaths_to_confirmed) if ratio_deaths_to_confirmed.finite?
new_event.set("ratio_recovered_to_confirmed", ratio_recovered_to_confirmed) if ratio_recovered_to_confirmed.finite?
new_event.set("ratio_deaths_to_recovered", ratio_deaths_to_recovered) if ratio_deaths_to_recovered.finite?

if new_event.get("province_state")&.match?(/(.*?),\s(\w{2})/)
matched = new_event.get("province_state").match(/(.*?),\s(\w{2})/)
@states[matched[2]].nil? ? new_event.set("province_state", matched[2]) : new_event.set("province_state", @states[matched[2]])
new_event.set("admin2", matched[1].gsub(/ [Cc]ounty/, ""))
new_event.tag("added_county")
end

#..generate a unique key for the current event
key = ((new_event.get("admin2") || "").downcase + "_" + (new_event.get("province_state") || "").downcase + "_" + (new_event.get("country_region") || "").downcase + "_" + (current_date.strftime("%m-%d-%Y") || "").to_s).gsub(/[^\w\d]/, "_")
new_event.set("unique_id", key)

#..add a tag to identify the event later
new_event.tag("covid_time_series")
events_array.append(new_event)

if ["Channel Islands", "Cayman Islands"].include?(new_event.get("country_region"))
new_event.set("province_state", new_event.get("country_region"))
new_event.set("country_region", "United Kingdom")
end
if ["Saint Martin", "St. Martin"].include?(new_event.get("country_region"))
new_event.set("province_state", "St Martin")
new_event.set("country_region", "France")
end

country_normalization = {
"Mainland China" => "China",
"South Korea" => "Korea, South",
"Republic of Korea" => "Korea, South",
"Iran \(Islamic Republic of\)" => "Iran",
"Bahamas, The" => "Bahamas",
"Gambia, The" => "Gambia",
"Hong Kong SAR" => "China",
"Hong Kong" => "China",
"Macao SAR" => "China",
"Macau" => "China",
"Republic of Ireland" => "Ireland",
"Republic of Moldova" => "Moldova",
"Republic of the Congo" => "Congo (Brazzaville)",
"Russian Federation" => "Russia",
"Taipei and environs" => "Taiwan",
"Taiwan\*" => "Taiwan",
"UK" => "United Kingdom",
"Vatican City" => "Holy See",
"Viet Nam" => "Vietnam",
"occupied Palestinian territory" => "West Bank and Gaza",
"Ivory Coast" => "Cote d\'Ivoire",
"East Timor" => "Timor-Leste",
"Czech Republic" => "Czechia",
"The Bahamas" => "Bahamas",
"The Gambia" => "Gambia",
"Cape Verde" => "Cabo Verde",
"North Ireland" => "Ireland",
"Palestine" => "West Bank and Gaza",
}
new_event.set("country_region", new_event.get("country_region").gsub(/.*/, country_normalization)) unless country_normalization[new_event.get("country_region")].nil?

if totals_by_country[new_event.get("country_region")].nil?
totals_by_country[new_event.get("country_region")] = LogStash::Event.new
totals_by_country[new_event.get("country_region")].set("confirmed", 0)
totals_by_country[new_event.get("country_region")].set("recovered", 0)
totals_by_country[new_event.get("country_region")].set("deaths", 0)
totals_by_country[new_event.get("country_region")].set("country_region", new_event.get("country_region"))
totals_by_country[new_event.get("country_region")].set("timestamp", new_event.get("timestamp"))
totals_by_country[new_event.get("country_region")].tag("total_by_country_region")
totals_by_country[new_event.get("country_region")].tag("covid_time_series")
totals_by_country[new_event.get("country_region")].set("unique_id", "totals_for_" + new_event.get("country_region").downcase + "_" + current_date.strftime("%m-%d-%Y").to_s.gsub(/[^\w\d]/, "_"))
end
totals_by_country[new_event.get("country_region")].set("confirmed", new_event.get("confirmed").to_i + totals_by_country[new_event.get("country_region")].get("confirmed").to_i)
totals_by_country[new_event.get("country_region")].set("deaths", new_event.get("deaths").to_i + totals_by_country[new_event.get("country_region")].get("deaths").to_i)
totals_by_country[new_event.get("country_region")].set("recovered", new_event.get("recovered").to_i + totals_by_country[new_event.get("country_region")].get("recovered").to_i)
ratio_deaths_to_confirmed = totals_by_country[new_event.get("country_region")].get("deaths").to_f / totals_by_country[new_event.get("country_region")].get("confirmed").to_f
ratio_recovered_to_confirmed = totals_by_country[new_event.get("country_region")].get("recovered").to_f / totals_by_country[new_event.get("country_region")].get("confirmed").to_f
ratio_deaths_to_recovered = totals_by_country[new_event.get("country_region")].get("deaths").to_f / totals_by_country[new_event.get("country_region")].get("recovered").to_f
totals_by_country[new_event.get("country_region")].set("ratio_deaths_to_confirmed", ratio_deaths_to_confirmed) if ratio_deaths_to_confirmed.finite?
totals_by_country[new_event.get("country_region")].set("ratio_recovered_to_confirmed", ratio_recovered_to_confirmed) if ratio_recovered_to_confirmed.finite?
totals_by_country[new_event.get("country_region")].set("ratio_deaths_to_recovered", ratio_deaths_to_recovered) if ratio_deaths_to_recovered.finite?

totals_by_country["WORLD"].set("total_confirmed", totals_by_country["WORLD"].get("total_confirmed").to_i + new_event.get("confirmed").to_i)
totals_by_country["WORLD"].set("total_recovered", totals_by_country["WORLD"].get("total_recovered").to_i + new_event.get("recovered").to_i)
totals_by_country["WORLD"].set("total_deaths", totals_by_country["WORLD"].get("total_deaths").to_i + new_event.get("deaths").to_i)
all_country_region.append(new_event.get("country_region")) if !new_event.get("country_region").nil? and !all_country_region.include?(new_event.get("country_region"))
all_province_state.append(new_event.get("province_state")) if !new_event.get("province_state").nil? and !all_province_state.include?(new_event.get("province_state"))
all_county.append(new_event.get("admin2")) if !new_event.get("admin2").nil? and !all_county.include?(new_event.get("admin2"))
end

ratio_deaths_to_confirmed = totals_by_country["WORLD"].get("total_deaths").to_f / totals_by_country["WORLD"].get("total_confirmed").to_f
ratio_recovered_to_confirmed = totals_by_country["WORLD"].get("total_recovered").to_f / totals_by_country["WORLD"].get("total_confirmed").to_f
ratio_deaths_to_recovered = totals_by_country["WORLD"].get("total_deaths").to_f / totals_by_country["WORLD"].get("total_recovered").to_f
totals_by_country["WORLD"].set("ratio_deaths_to_confirmed", ratio_deaths_to_confirmed) if ratio_deaths_to_confirmed.finite?
totals_by_country["WORLD"].set("ratio_recovered_to_confirmed", ratio_recovered_to_confirmed) if ratio_recovered_to_confirmed.finite?
totals_by_country["WORLD"].set("ratio_deaths_to_recovered", ratio_deaths_to_recovered) if ratio_deaths_to_recovered.finite?
totals_by_country["WORLD"].set("country_region", all_country_region)
totals_by_country["WORLD"].set("province_state", all_province_state)
totals_by_country["WORLD"].set("admin2", all_county)
totals_by_country.each do |k, v|
events_array.append(v)
end

return events_array
end
'
code => '
#First day in the dataset
github_daily_reports_first_day = Time.gm(2020, 01, 22, 12, 00)
#Current day
github_daily_reports_last_day = Time.now
current_date = github_daily_reports_first_day

#If the CSV hashes are stored in a file, parse them..
if File.file?(@github_stored_hashes_path)
stored_hashes = JSON.parse(File.read(@github_stored_hashes_path))
#..else, create a new hash
else
stored_hashes = Hash.new
end

#For each day since the beggining of the dataset, until the current day..
while current_date <= github_daily_reports_last_day="" begin="" #..generate="" the="" url="" based="" on="" current_day="" current_url="@github_daily_reports_base_url" +="" current_date.strftime("%m-%d-%y")="" ".csv"="" #..parse="" csv="" while="" normalizing="" headers="" #(e.g.="" until="" 21.03.2020,="" country="" is="" stored="" under="" "country="" region".="" after="" this="" date,="" it="" "country_region"="" #we="" normalize="" such="" that="" both="" variations="" are="" ultimately="" "country_region".="" applies="" to="" other="" fields="" as="" well)="" current_csv="CSV.parse(open(current_url).read.gsub(" \ufeff","="" ""),="" headers:="" true,="" header_converters:="" lambda="" {="" |h|="" h.downcase.gsub(="" [^\w]="" ,="" "_")="" })="" hash="" current_md5="Digest::SHA256.hexdigest(current_csv.to_s)" rescue="" #the="" pipeline="" will="" usually="" fail="" fetch="" data="" for="" current="" day,="" if="" was="" not="" already="" published="" logger.warn?="" and="" logger.warn("failed="" date="" "="" current_date.strftime("%m-%d-%y"))="" else="" #..if="" different="" than="" one,="" process="" events="" stored_hashes[current_url]="" !="current_md5" events_array="create_events(current_csv," current_date)="" #..store="" new="" #..and="" send="" each="" event="" back="" through="" events_array.each="" do="" |event|="" new_event_block.call(event)="" end="" #check="" next="" day="" current_date="" #attempt="" write="" hashes="" disk="" file.write(@github_stored_hashes_path,="" stored_hashes.to_json)="" logger.info?="" logger.info("succesfuly="" updated="" cache="" at="" #{@github_stored_hashes_path}")="" standarderror=""> ex</=>
logger.error? and logger.error("(#{ex.class} - #{ex.message}) Failed to write hashes to file #{@github_stored_hashes_path}. Does Logstash have read/write access to this location?")
ensure
#..and cancel this event. We are only interested in the time series data, which was already sent back through the pipeline.
event.cancel
end
'
}#end ruby
}#end if

#Each time series data event will be sent back through the pipeline.
#This 'if' discriminates between the original event that triggered the downloading and processing of the CSV, and the time series data
if "covid_time_series" in [tags] {
#Parse date fields
date {
#When the github dataset was last updated by the maintainers
match => [ "last_update", "yyyy-MM-dd HH:mm:ss", "M/d/yy HH:mm", "M/d/yyyy HH:mm", "ISO8601" ]
target => "last_update"
}#end date
date {
#When the event was processed by Logstash
match => ["last_process", "ISO8601" ]
target => "last_process"
}#end date
date {
#The true date of the report
match => [ "timestamp", "ISO8601" ]
target => "@timestamp"
}#end date

if "total_by_country_region" not in [tags] and "total_world" not in [tags] {
mutate {
#Normalize the location field
rename => {
"lat" => "[location][lat]"
"latitude" => "[location][lat]"
"lon" => "[location][lon]"
"long" => "[location][lon]"
"long_" => "[location][lon]"
"longitude" => "[location][lon]"
}
}
ruby {
code => '
@state_coordinates = { "Wisconsin" => { "lat" => "44.5", "lon" => "-89.5" }, "West Virginia" => { "lat" => "39", "lon" => "-80.5" }, "Vermont" => { "lat" => "44", "lon" => "-72.699997" }, "Texas" => { "lat" => "31", "lon" => "-100" }, "South Dakota" => { "lat" => "44.5", "lon" => "-100" }, "Rhode Island" => { "lat" => "41.700001", "lon" => "-71.5" }, "Oregon" => { "lat" => "44", "lon" => "-120.5" }, "New York" => { "lat" => "43", "lon" => "-75" }, "New Hampshire" => { "lat" => "44", "lon" => "-71.5" }, "Nebraska" => { "lat" => "41.5", "lon" => "-100" }, "Kansas" => { "lat" => "38.5", "lon" => "-98" }, "Mississippi" => { "lat" => "33", "lon" => "-90" }, "Illinois" => { "lat" => "40", "lon" => "-89" }, "Delaware" => { "lat" => "39", "lon" => "-75.5" }, "Connecticut" => { "lat" => "41.599998", "lon" => "-72.699997" }, "Arkansas" => { "lat" => "34.799999", "lon" => "-92.199997" }, "Indiana" => { "lat" => "40.273502", "lon" => "-86.126976" }, "Missouri" => { "lat" => "38.573936", "lon" => "-92.60376" }, "Florida" => { "lat" => "27.994402", "lon" => "-81.760254" }, "Nevada" => { "lat" => "39.876019", "lon" => "-117.224121" }, "Maine" => { "lat" => "45.367584", "lon" => "-68.972168" }, "Michigan" => { "lat" => "44.182205", "lon" => "-84.506836" }, "Georgia" => { "lat" => "33.247875", "lon" => "-83.441162" }, "Hawaii" => { "lat" => "19.741755", "lon" => "-155.844437" }, "Alaska" => { "lat" => "66.160507", "lon" => "-153.369141" }, "Tennessee" => { "lat" => "35.860119", "lon" => "-86.660156" }, "Virginia" => { "lat" => "37.926868", "lon" => "-78.024902" }, "New Jersey" => { "lat" => "39.833851", "lon" => "-74.871826" }, "Kentucky" => { "lat" => "37.839333", "lon" => "-84.27002" }, "North Dakota" => { "lat" => "47.650589", "lon" => "-100.437012" }, "Minnesota" => { "lat" => "46.39241", "lon" => "-94.63623" }, "Oklahoma" => { "lat" => "36.084621", "lon" => "-96.921387" }, "Montana" => { "lat" => "46.96526", "lon" => "-109.533691" }, "Washington" => { "lat" => "47.751076", "lon" => "-120.740135" }, "Utah" => { "lat" => "39.41922", "lon" => "-111.950684" }, "Colorado" => { "lat" => "39.113014", "lon" => "-105.358887" }, "Ohio" => { "lat" => "40.367474", "lon" => "-82.996216" }, "Alabama" => { "lat" => "32.31823", "lon" => "-86.902298" }, "Iowa" => { "lat" => "42.032974", "lon" => "-93.581543" }, "New Mexico" => { "lat" => "34.307144", "lon" => "-106.018066" }, "South Carolina" => { "lat" => "33.836082", "lon" => "-81.163727" }, "Pennsylvania" => { "lat" => "41.203323", "lon" => "-77.194527" }, "Arizona" => { "lat" => "34.048927", "lon" => "-111.093735" }, "Maryland" => { "lat" => "39.045753", "lon" => "-76.641273" }, "Massachusetts" => { "lat" => "42.407211", "lon" => "-71.382439" }, "California" => { "lat" => "36.778259", "lon" => "-119.417931" }, "Idaho" => { "lat" => "44.068203", "lon" => "-114.742043" }, "Wyoming" => { "lat" => "43.07597", "lon" => "-107.290283" }, "North Carolina" => { "lat" => "35.782169", "lon" => "-80.793457" }, "Louisiana" => { "lat" => "30.39183", "lon" => "-92.329102" } }

if (event.get("[location][lat]").nil? || event.get("[location][lon]").nil?) || (event.get("[location][lat]").to_f == 0 && event.get("[location][lon]").to_f == 0)
if (event.get("[country_region]") == "US") and (!@state_coordinates[event.get("[province_state]")].nil?)
event.set("[location][lat]", @state_coordinates[event.get("province_state")]["lat"])
event.set("[location][lon]", @state_coordinates[event.get("province_state")]["lon"])
else
event.set("[location][lat]", 0)
event.set("[location][lon]", 0)
end
end
'
}
}

#Rename the fields if you want to correlate with other datasets
mutate {
rename => {
"province_state" => "province_state"
"country_region" => "country_region"
"admin2" => "county"
"fips" => "fips"

"confirmed" => "confirmed"
"deaths" => "deaths"
"recovered" => "recovered"
"active" => "active"

"[location][lat]" => "[location][lat]"
"[location][lon]" => "[location][lon]"

"last_update" => "last_update"
"last_process" => "last_process"
"combined_key" => "combined_key"

"ratio_deaths_to_confirmed" => "ratio_deaths_to_confirmed"
"ratio_recovered_to_confirmed" => "ratio_recovered_to_confirmed"
"ratio_deaths_to_recovered" => "ratio_deaths_to_recovered"

"unique_id" => "unique_id"
}
}#end mutate
}#end if
}#end filter

output {
#Send the data to Elasticsearch
elasticsearch {
#Add your Elasticsearch hosts
hosts => [""]
#Add the index name
index => "covid-19-live-update"
#Add Elasticsearch credentials
user => ""
password => ""
#Add SSL details
#ssl => true
#cacert => "/path/to/cert/"

#Update documents based on the unique id that we defined
action => "update"
document_id => "%{unique_id}"
#..or create a new document if it doesn't already exist
doc_as_upsert => true
manage_template => false
}
}

Elasticsearch index

In addition to the Logstash code you will also require a template for the index in elasticsearch. The field mapping can be found below (feel free to change the name of the index but remember to change it also in the Logstash code):

-- CODE markdown--
PUT _template/covid-19-live-update
{
"order" : 0,
"index_patterns" : [
"covid-19-live-update"
],
"settings" : { },
"mappings" : {
"properties" : {
"recovered" : {
"type" : "long"
},
"updated_at" : {
"type" : "date"
},
"ratio_deaths_to_recovered" : {
"type" : "float"
},
"ratio_recovered_to_confirmed" : {
"type" : "float"
},
"ratio_deaths_to_confirmed" : {
"type" : "float"
},
"location" : {
"type" : "geo_point"
},
"confirmed" : {
"type" : "long"
},
"deaths" : {
"type" : "long"
},
"last_update" : {
"type" : "date"
},
"last_process" : {
"type" : "date"
},
"@timestamp" : {
"type" : "date"
}
}
},
"aliases" : { }
}

Kibana dashboard

In the GitHub repository you will also find the dashboard from this article which you can import into your own Kibana. The dashboard was built using the latest version of the Elastic Stack (7.6.1). In order to import the dashboard you will have to navigate in the Kibana interface to Management > Saved Objects and click import.

The GitHub repository can be found here.

Watcher alerting

Once these steps are done you will have the data into Elasticsearch and available through Kibana.

We are also attaching two Watcher codes (alerting) to this article which will send an email to the configured email address with details regarding the number of infected people, recovered and deceased depending on the region of interest. Because not all countries have information at the province/region level (this is how the data is collected by John Hopkins Hospital) there is one Watcher for regional data and one for country-wide data. In the code you will find the following tag “change this” which should be changed with the desired value.

A few considerations for the alerts:

Make sure to change just the text within the double quotes without deleting them
Province and country names should be written with capital letters ( e.g. New York, China etc.)
Feel free to change the interval value at the top of the code with the desired interval with which you want the alerting to occur (values are in the form of a number followed by m for minute, h for hour, d for days without deleting the double quotes)
Keep in mind that the data from John Hopkins Hospital is updated once or twice a day from our observations
The following countries have regional information:
US
China
Australia
Canada
France
United Kingdom
Denmark
Cruise Ship
In order to be able to send emails you need to configure your Elastic Stack accordingly. Details can be found here.

Province/Region Watcher alert

-- CODE markdown--
{
"trigger": {
"schedule": {
"interval": "24h"
}
},
"input": {
"search": {
"request": {
"body": {
"query" : {
"bool" : {
"must" : [
{
"match" : {
"province_state.keyword" : "Change this"
}
}
]
}
},
"size" : 2,
"sort" : {
"@timestamp" : {
"order" : "desc"
}
}
},
"indices": [
"covid-19-live-update"
]
}
}
},
"condition": {
"compare" : {
"ctx.payload.hits.total" : {
"gt" : 1
}
}
},
"transform" : {
"script" : "return [ 'country' : ctx.payload.hits.hits[0]._source.country_region, 'province' : ctx.payload.hits.hits[0]._source.province_state, 'confirmed_today' : ctx.payload.hits.hits[0]._source.confirmed, 'deaths_today' : ctx.payload.hits.hits[0]._source.deaths, 'recovered_today' : ctx.payload.hits.hits[0]._source.recovered, 'confirmed_yesterday' : ctx.payload.hits.hits[1]._source.confirmed, 'deaths_yesterday' : ctx.payload.hits.hits[1]._source.deaths, 'recovered_yesterday' : ctx.payload.hits.hits[1]._source.recovered, 'difference_confirmed' : (Integer.parseInt(ctx.payload.hits.hits[0]._source.confirmed) - Integer.parseInt(ctx.payload.hits.hits[1]._source.confirmed)), 'difference_deaths' : (Integer.parseInt(ctx.payload.hits.hits[0]._source.deaths) - Integer.parseInt(ctx.payload.hits.hits[1]._source.deaths)), 'difference_recovered' : (Integer.parseInt(ctx.payload.hits.hits[0]._source.recovered) - Integer.parseInt(ctx.payload.hits.hits[1]._source.recovered))]"
},
"actions": {
"email" : {
"email" : {
"to" : "username@example.org",
"subject" : "COVID-19 Status for {{ctx.payload.province}}, {{ctx.payload.country}}",
"body" : "Confirmed/Recovered/Dead: {{ctx.payload.confirmed_today}} (+{{ctx.payload.difference_confirmed}}) / {{ctx.payload.recovered_today}} (+{{ctx.payload.difference_recovered}}) / {{ctx.payload.deaths_today}} (+{{ctx.payload.difference_deaths}})"
}
}
}
}

Country level Watcher alert

-- CODE markdown--
{
"trigger": {
"schedule": {
"interval": "24h"
}
},
"input": {
"search": {
"request": {
"body": {
"query" : {
"bool" : {
"must" : [
{
"match" : {
"country_region.keyword" : "Change this"
}
}
]
}
},
"size" : 2,
"sort" : {
"@timestamp" : {
"order" : "desc"
}
}
},
"indices": [
"covid-19-live-update"
]
}
}
},
"condition": {
"compare" : {
"ctx.payload.hits.total" : {
"gt" : 1
}
}
},
"transform" : {
"script" : "return [ 'country' : ctx.payload.hits.hits[0]._source.country_region, 'province' : ctx.payload.hits.hits[0]._source.province_state, 'confirmed_today' : ctx.payload.hits.hits[0]._source.confirmed, 'deaths_today' : ctx.payload.hits.hits[0]._source.deaths, 'recovered_today' : ctx.payload.hits.hits[0]._source.recovered, 'confirmed_yesterday' : ctx.payload.hits.hits[1]._source.confirmed, 'deaths_yesterday' : ctx.payload.hits.hits[1]._source.deaths, 'recovered_yesterday' : ctx.payload.hits.hits[1]._source.recovered, 'difference_confirmed' : (Integer.parseInt(ctx.payload.hits.hits[0]._source.confirmed) - Integer.parseInt(ctx.payload.hits.hits[1]._source.confirmed)), 'difference_deaths' : (Integer.parseInt(ctx.payload.hits.hits[0]._source.deaths) - Integer.parseInt(ctx.payload.hits.hits[1]._source.deaths)), 'difference_recovered' : (Integer.parseInt(ctx.payload.hits.hits[0]._source.recovered) - Integer.parseInt(ctx.payload.hits.hits[1]._source.recovered))]"
},
"actions": {
"email" : {
"email" : {
"to" : "username@example.org",
"subject" : "COVID-19 Status for {{ctx.payload.country}}",
"body" : "Confirmed/Recovered/Dead: {{ctx.payload.confirmed_today}} (+{{ctx.payload.difference_confirmed}}) / {{ctx.payload.recovered_today}} (+{{ctx.payload.difference_recovered}}) / {{ctx.payload.deaths_today}} (+{{ctx.payload.difference_deaths}})"
}
}
}
}

In part 2 of this blog post we will also show how you can correlate the Coronavirus data with other data sets. This will help businesses to correlate their operational and business data with the COVID-19 outbreak in order to adjust their activities.

If you require help or assistance with this process please contact us at contact@siscale.com