Robin on Linux – Page 17 – All about technology

A problem of using Pyspark SQL

Here is the code:

from pyspark.sql import SQLContext
from pyspark.context import SparkContext
from pyspark.sql.types import *
from typing import List
sc = SparkContext()
sqlContext = SQLContext.getOrCreate(sc)
schema = StructType([StructField('id', LongType(), True),
                      StructField('gid', LongType(), True),
                      StructField('pid', LongType(), True),
                      StructField('firstlogin', IntegerType(), True)
])
row = ['2', '29', '29', '29']
df = sqlContext.createDataFrame(row, schema)
df.show()

It will report error after running ‘cat xxx.py|bin/pyspark’:

TypeError: StructType can not accept object '2' in type

I used to think it was because ‘2’ is a string, so I changed ‘row’ to be ‘[2, 29, 29, 29]’. But the error also changed to:

TypeError: StructType can not accept object 2 in type

Then I searched on google, and find this article. Looks like I forgot to transfer ‘list’ of python to ‘RDD’ of Apache Spark.
But at last, I found the real reason: I just need to add ‘[]’ between my ‘list’!
The right code is here:

row = ['2', '29', '29', '29']
df = sqlContext.createDataFrame([row], schema)

Some problems about using AWS DMS

AWS DMS is a new type of service used to migrate data from different types of database and data-warehouse. I met some problems when trying to use it in production environment.
Problem 1. When using a MySQL server of AWS RDS as the source of a replication task. It reported errors after started the task:

Last failure message
Last Error Task error notification received from subtask 0, thread 0 [reptask/replicationtask.c:2673] [1020418] Error Code [10001] : Binary Logging must be enabled for MySQL server; Errors in MySQL server binary logging configuration. Follow all prerequisites for 'MySQL as a source in DMS' from https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Source.MySQL.html or'MySQL as a target in DMS' from https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.MySQL.html ; Failed while preparing stream component 'st_0_WBK5KGUWQAH6VKEP4I5LH2EFHE'.; Cannot initialize subtask; Stream component 'st_0_WBK5KGUWQAH6VKEP4I5LH2EFHE' terminated [reptask/replicationtask.c:2680] [1020418] Stop Reason FATAL_ERROR Error Level FATAL

The failure message looks terrible. But at least I can find this doc to follow. After changed the configurations as below:

binlog_format	ROW
binlog_checksum	NONE
binlog_row_image	FULL

the error still existed.
The real answer is in here since I used RDS instead of self-managed MySQL. After I add one line Terraform code to enable “automatic backups”:

resource "aws_db_instance" "test_gaf" {
  ......
  backup_retention_period     = 10
}

the replication task began to work without the error.
Problem 2. Running replication task for a while to export data from MySQL to AWS Redshift. A new error log appeared in Redshift load logs:

019-10-29T04:41:27 [TARGET_LOAD ]E: RetCode: SQL_ERROR SqlState: XX000 NativeError: 30 Message: [Amazon][Amazon Redshift] (30) Error occurred while trying to execute a query: [SQLState XX000] ERROR: User arn:aws:redshift:us-east-1:262284277472:dbuser:analytics-20190902/masteruser is not authorized to assume IAM Role arn:aws:iam::262284277472:role/dms-access-for-endpoint DETAIL: ----------------------------------------------- error: User arn:aws:redshift:us-east-1:262284277472:dbuser:analytics-20190902/masteruser is not authorized to assume IAM Role arn:aws:iam::262284277472:role/dms-access-for-endpoint code: 8001 context: IAM Role=arn:aws:iam::262284277472:role/dms-access-for-endpoint query: 1799 location: xen_aws_credentials_mgr.cpp:321 process: padbmaster [pid=21755] ----------------------------------------------- [1022502] (ar_odbc_stmt.c:4622)

Why masteruser is not authorized? The answer is here. Below is the Terraform code:

data "aws_iam_policy_document" "dms_assume_role" {
  statement {
    actions = ["sts:AssumeRole"]
    principals {
      identifiers = ["dms.amazonaws.com"]
      type        = "Service"
    }
  }
  statement {
    actions = ["sts:AssumeRole"]
    # By https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Security.APIRole.html,
    # we also need principal `redshift.amazonaws.com`
    principals {
      identifiers = ["redshift.amazonaws.com"]
      type        = "Service"
    }
  }
}

Then I had giiven “dms_assume_role” two Trusty Entities

Problem 3. There was still a error in Redshift load log (so many errors in AWS DMS…):

Error	Type	Raw Field Value
Invalid timestamp format or value [YYYY-MM-DD HH24:MI:SS]	timestamp	0000-00-00 00:00:00

Seems the answer is here. Therefore I added “acceptanydate=true;timeformat=auto” into the “extra connection settings” in Redshift endpoint. But the error just changed to:

Error	Type	Raw Field Value
Invalid data	timestamp	0000-00-00 00:00:00

After searching for almost two days, I found that the reason is in the schema of Redshift, which is automatically created by AWS DMS replication task.

CREATE TABLE my (
    ...
    mydate TIMESTAMP DEFAULT '0000-00-00 00:00:00' NOT NULL,
    ...
)

Since the schema doesn’t allow “mydate” column to be null but the “acceptanydate=true” is trying to transfer “0000-00-00 00:00:00 to null”, the final error is “Invalid data” for Redshift.
The solution for this problem is: create table of Redshift manually to let “mydate” column to be “nullable”, and change the working mode of replication task to “TRUNCATE_BEFORE_LOAD”.

Processing date and time in AWS Redshift

Since AWS Redshift don’t have function like FROM_UNIX(), it’s much more weird to get formatted time from a UNIX timestamp (called ‘epoch’ in Reshift):

SELECT timestamp 'epoch' + my_timestamp_column * interval '1 second' AS my_column_alias
FROM my_table

Ref: https://stackoverflow.com/questions/39815425/how-to-convert-epoch-to-datetime-redshift
If we want to see the statistics result group by hours:

SELECT
  COUNT(1),
  DATE_TRUNC('hour', timestamp 'epoch' + my.timestamp * interval '1 second') AS hour
FROM my_table AS my
WHERE my_table.my_timestamp_column > extract('epoch' from '2019-10-23'::TIMESTAMP)
GROUP BY hour
ORDER BY hour;

Some tips about using AWS Glue

Configure about data format
To use AWS Glue, I write a ‘catalog table’ into my Terraform script:

resource "aws_glue_catalog_table" "my_table" {
...
    input_format  = "org.apache.hadoop.mapred.TextInputFormat"
    output_format = "org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat"
    ser_de_info {
      name                  = "SerDeCsv"
      serialization_library = "org.apache.hadoop.hive.serde2.OpenCSVSerde"
      parameters = {
        "separatorChar" = ","
        "quoteChar"     = "'"
      }
    }
...
}

But after using PySpark script to access this table, it reports:

py4j.protocol.Py4JJavaError: An error occurred while calling o58.getCatalogSource.
: com.amazonaws.services.glue.util.NonFatalException: Formats not supported for SparkSQL data sources. Got csv
	at com.amazonaws.services.glue.SparkSQLDataSource.setFormat(DataSource.scala:641)
	at com.amazonaws.services.glue.GlueContext.getCatalogSource(GlueContext.scala:254)
	at com.amazonaws.services.glue.GlueContext.getCatalogSource(GlueContext.scala:139)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
	at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
	at py4j.Gateway.invoke(Gateway.java:280)
	at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
	at py4j.commands.CallCommand.execute(CallCommand.java:79)
	at py4j.GatewayConnection.run(GatewayConnection.java:214)
	at java.lang.Thread.run(Thread.java:748)

Seems we can’t use ‘OpenCSVSerde’. Actually, the correct answer is:

Input format: org.apache.hadoop.mapred.TextInputFormat
Output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
Serde serialization lib: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
Serde parameters: field.delim ,

The version of zeppelin
When using zeppelin to run PySpark script, it reports error:

org.apache.thrift.TApplicationException: Internal error processing createInterpreter
	at org.apache.thrift.TApplicationException.read(TApplicationException.java:111)
	at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:71)
	at org.apache.zeppelin.interpreter.thrift.RemoteInterpreterService$Client.recv_createInterpreter(RemoteInterpreterService.java:209)
	at org.apache.zeppelin.interpreter.thrift.RemoteInterpreterService$Client.createInterpreter(RemoteInterpreterService.java:192)
	at org.apache.zeppelin.interpreter.remote.RemoteInterpreter$2.call(RemoteInterpreter.java:169)
	at org.apache.zeppelin.interpreter.remote.RemoteInterpreter$2.call(RemoteInterpreter.java:165)
	at org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess.callRemoteFunction(RemoteInterpreterProcess.java:135)
	at org.apache.zeppelin.interpreter.remote.RemoteInterpreter.internal_create(RemoteInterpreter.java:165)
	at org.apache.zeppelin.interpreter.remote.RemoteInterpreter.open(RemoteInterpreter.java:132)
	at org.apache.zeppelin.interpreter.remote.RemoteInterpreter.getFormType(RemoteInterpreter.java:299)
	at org.apache.zeppelin.notebook.Paragraph.jobRun(Paragraph.java:407)
	at org.apache.zeppelin.scheduler.Job.run(Job.java:188)
	at org.apache.zeppelin.scheduler.RemoteScheduler$JobRunner.run(RemoteScheduler.java:315)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

According to the document:

The latest release of Apache Zeppelin, 0.8.x, is not supported. Download the older release named zeppelin-0.7.3-bin-all.tgz from the download page and follow the installation instructions.

Google Cloud Summit 2019

Yesterday I joined the Google Cloud Summit 2019 in Sydney.
The meeting place is quite huge. And there are lot of booths from different partners of Google Cloud.

The keynote was quite abstract and a little boring, so I chose to walk around different booths for fun. Here are some useful conversations and information I collected:
[HashiCorp: company for Terraform and Nomad]
Q: How short could Terraform support a new service in cloud provider, such as Lake Formation in AWS?
A: It depends on requirements from users.
Q: Only users who paid for enterprise version of Terraform?
A: No. All users, include who use open source version, will be considered to support a new service in cloud. We published new version of Terraform quarterly, although we can’t make sure every new services in this quarter will be included.
[Confluent: company for Apache Kafka]
Q: Could I use ksql in Kafka as standard SQL?
A: Currently, no. ksql only supports self-defined syntax in Kafka. It looks really like SQL, but it’s actually another language.
Q: Could I use ksql to access a table in MySQL?
A: Yes. You can export a table in MySQL to be a ‘kstream‘. Then you can use ksql to access this ‘kstream’.
[Tableau: you know what I mean…]
Q: What are the new updates for Tableau in recent one year?
A: We published a new function called ‘Ask Data’. You can type in query with natural language, and it will translate them to tableau query, by using state-of-the-art NLP technologies.
After I type in some query like ‘avg price in Manly’, it worked very well. But if I type in query like ‘top 5 price near Chatswood’, Tableau failed to get the right answer.
A: You know, NLP is really hard so we only support a range of anonymous for query words.
[elastic: company for ElasticSearch and Kibana]
Q: What’s the biggest cluster of ElasticSearch in production?
A: Well, a lot of big companies use quit big ElasticSearch clusters, such as Netflix, eBay. But we don’t know which one is the biggest because they won’t tell us every details of their clusters 🙂
[Google Cloud]
Q: Is there a product in Google Cloud that could continually import data from MySQL and export them to Cloud Storage or BigTable?
A: Yes. Cloud Data Fusion will be your best choice.
In the booth of ASUS (it produced a lot of chrome books for Google), I noticed the Dev Board which contain a edge-TPU.

The demo use “mobilenet_ssd v2” as the backbone for object detection. Just as my choice.

Accelerate the speed of data loading in PyTorch

I got a desktop computer to train deep learning model last week. The GPU is GTX1050TI with 4GB memory which is enough for basic training on object detection. But the CPU is too old. Therefore when I run the training process, the idle of CPU is 0%. I need to reduce the burden of it.
I tried DALI from Nvidia. Although confessed that it is powerful, I also noticed that DALI is too specific to be used for customer dataset. For example, if I want to use complicate label structures more than just ‘bounding box’ coordinates, I can’t find any code example to in DALI to meet this requirement. By the way, the GPU memory in my computer is not big enough, so if moved computation burden from CPU to GPU, I would have to reduce batch size for training. That’s not a good option too.
Yesterday, from this post, I saw this suggestion:

You can use jpeg4py, library dedicated to encode big jpeg files much faster than PIL. Just read image using this library, then transform it to PIL.

After changed my code from using ‘cv.imread()’ to jpeg4py.JPEG().decode()’, the average training time for 1000 batches in my mode boosted improved 700 seconds to 670 seconds. This is just the simple and useful solution I need.

The uneasy way to implement SSDLite by myself

SSDLite is a variant of Single Shot Multi-box Detection. It uses MobileNetV2 instead of VGG as backbone. Thus it can make detection extremely fast. I was trying to implement SSDLite from the code base of ssd.pytorch. Although it’s not a easy work, I finally learn a lot from the entire process.
First, I just replace VGG with MobileNetV2 in the code. However the loss will stop to reduce after a while of training. Without knowing the reason, I have to compare my code with another open source project ssds.pytorch, to try to find the cause.
Very soon, I noticed that unlike VGG backbone, which built detection framework from 38×38 feature map, the MobileNetV2 use 19×19 feature map as its first detection layer.

“For MobileNetV1, we follow the setup in [33]. For MobileNetV2, the first layer of SSDLite is attached to the expansion of layer 15 (with output stride of 16).”
From: MobileNetV2: Inverted Residuals and Linear Bottlenecks

After changed my code as the description of this paper, the loss still couldn’t reduce in training.
In the next three weeks, I tried a lot of methods: change the aspect ratios, use SGDR to replace SGD, change the number of default boxes, even modifying the structure of neural network to be identical to to ssds.pytorch. But none of them solves the problem. There is another weird phenomenon: when I run prediction on my model, it usually gives random output for detection.
Just until last week, I noticed that my model size is about 10MB but the ssds.pytorch’s is 18MB. Why do they have different model size if their models is exactly the same? Through this clue, I eventually get the cause: a large part of my model hasn’t been back-propagated at all!
My old code only implements the forward() of MobileNetV2 which is not enough for the whole model. Therefore I add nn.ModuleList() to build model from a list of layers, as this patch:

self.backbone = nn.ModuleList(mobilenet.MobileNetV2(num_classes=num_classes, width_mult=1.0).features)

Only the nn.ModuleList() will take all layers into back-propagation process and keep them as model weights. Otherwise, the weights will be randomly init and just use for forwarding — that’s why I get random output before.
I think I should be more carefully when adding FPN into my model in the future.

A tip about Terraform

Terraform is a interesting (in my opinion) tool to implement Infrastructure-as-Code. When I first used it to write production script at yesterday, I met a error report:

Error: error validating provider credentials: error calling sts:GetCallerIdentity: NoCredentialProviders: no valid providers in chain. Deprecated.
        For verbose messaging see aws.Config.CredentialsChainVerboseErrors

After a while of searching on Google, I got the cause: it can’t find my AWS credential in my computer.
Actually I have ‘~/.aws/credentials’ file, and the ‘access_key_id’, ‘secret_access_key’ are already in it. Like this:

[default]
aws_access_key_id = ABCDEFG
aws_secret_access_key = A1B2C3D4abcABC

So why can’t Terraform get the credential? The reason is in the ‘provider’ section:

provider "aws" {
  region = var.region
  profile = var.aws_profile
}

I set the ‘profile’ to ‘analytics’ at first, so the Terraform tried to find something looks like ‘[analytics]’ in ‘~/.aws/credentials’ file, and it failed. The correct way is just set ‘profile’ in ‘provider’ section to ‘default’.

Some ideas about building streaming ETL on AWS

After discussed with technical support guys from AWS, I get more information about how to use all the service of AWS to build a streaming ETL architecture, step by step.
The main architecture could be described by the diagram below:

AWS S3 is the de-facto data lake. All the data, no matter from AWS RDS or AWS Dynamo or other custom ways, could be written into AWS S3 by using some specific format, such as Apache Parquet or Apache ORC (CSV format is not recommend because it’s not suitable for data scan and data compression). Then, data engineers could use AWS Glue to extract the data from AWS S3, transform them (using PySpark or something like it), and load them into AWS Redshift.
For some frequently-used data, they could also be put in AWS Redshift for optimised query. When it is needed to join tables from both AWS S3 and AWS Redshift, we could also use AWS Redshift Spectrum.
BTW, I also joined a workshop about DataBricks’ new Unified Data Analytics and Machine Learning Platform which is built on AWS. It contains
1. Delta Lake for data storage and schema enforcement.
2. Notebook to let user directly write code and run them to process and analyze data by Apache Spark. Just like Jupyter Notebook.
3. MLFlow use above data to train machine learning model.
I used Apache Spark for learning about 4 years ago. At that time, I even need to build java/scala package by myself, upload and run it. Debugging is tedious because I can only scan logs of CLI again and again to find mistakes in code. But now, Databricks give a much more convenient solution for the data scientists and developers.

Someone who is interesting in this platform could try free edition of it https://databricks.com/try-databricks

The weird comparison behaviours of Python string

A part of my code didn’t work well, as below:

if mapping and mapping['colour'] is 'Red':
    print('Special colour')

It will not print out anything totally. So I directly printed out the actually value of mapping[‘colour’]:

print(mapping['colour'])
# It showed 'Red' in console

Why ‘Red’ is not ‘Red’? After changed the judgement from ‘is’ to ‘==’, the result became correct.
The key is that UNICODE string and normal string is different in Python:

u'Red' is 'Red' # False
u'Red' == 'Red' # True

Seems we should use ‘==’ to compare two strings instead of ‘is’.