Zeppelin 2019(1)Alarm System monitor DynamoDB

Zeppelin 2019(1)Alarm System monitor DynamoDB

First of all, I need shell interpreter there
%sh

cd ~/data
/opt/hadoop/bin/hdfs dfs -ls hdfs://localhost:9000/
CURRENT_DATE=`/bin/date +%Y-%m-%d`
DATA_PATH="monitor-$CURRENT_DATE"
echo $CURRENT_DATE
echo $DATA_PATH

Then I need to enable cron or scheduler as well.
https://zeppelin.apache.org/docs/0.8.1/usage/other_features/cron_scheduler.html#enable-cron
> vi conf/zeppelin-site.xml
<property>
  <name>zeppelin.notebook.cron.enable</name>
  <value>true</value>
  <description>Notebook enable cron scheduler feature</description>
</property>

After change the script, restart the application
> sudo bin/zeppelin-daemon.sh stop

> sudo bin/zeppelin-daemon.sh start

Change the timezone and restart zeppelin again
> sudo timedatectl set-timezone America/Chicago

Run the CRON job follow this format
0 22 00 * * ?  - every night 12:22:00

Shell command to dump the data

Extend the timeout time
shell.command.timeout.millisecs = 60000 by default, I changed that to few minutes.

Enable Variable
zeppelin.shell.interpolation = true

These commands are working fine in shell
%sh

CURRENT_DATE=`/bin/date +%Y-%m-%d`
DATA_PATH="monitor-$CURRENT_DATE"
echo $CURRENT_DATE
echo $DATA_PATH
/opt/hadoop/bin/hdfs dfs -mkdir hdfs://localhost:9000/$DATA_PATH
mkdir /home/ubuntu/data/$DATA_PATH
/home/ubuntu/.local/bin/export-dynamodb -t meeting-prod-meetings -f csv -o /home/ubuntu/data/$DATA_PATH/meetings.csv
/opt/hadoop/bin/hdfs dfs -put /home/ubuntu/data/$DATA_PATH/meetings.csv  hdfs://localhost:9000/$DATA_PATH/meetings.csv

I will use variable to short the command lines
val currentDate = java.time.LocalDate.now.toString()
z.put("CURRENT_DATE", currentDate)
val dataPath = "monitor-" + currentDate
z.put("DATA_PATH", dataPath)
val monitorLocalPath = "/home/ubuntu/data/" + dataPath
z.put("MONITOR_LOCAL_PATH", monitorLocalPath)
val monitorRemotePath = "hdfs://localhost:9000/" + dataPath
z.put("MONITOR_REMOTE_PATH", monitorRemotePath)
z.put("bin_hdfs", "/opt/hadoop/bin/hdfs")
z.put("bin_export-dynamodb", "/home/ubuntu/.local/bin/export-dynamodb")

So later, when we dump and load data, we can use short command
%sh

{bin_hdfs} dfs -mkdir {MONITOR_REMOTE_PATH}
mkdir {MONITOR_LOCAL_PATH}
{bin_export-dynamodb} -t meeting-prod-meetings -f csv -o {MONITOR_LOCAL_PATH}/meetings.csv
{bin_hdfs} dfs -put {MONITOR_LOCAL_PATH}/meetings.csv  {MONITOR_REMOTE_PATH}/meetings.csv


Send out Email
%spark.dep
z.load("com.sun.mail:javax.mail:1.6.2")

Check if Bad Data Exist
val badMeetingDF = sqlContext.sql("""
select
    UUID
from
    meetings
where
    clientApiID is null
    and ( from_unixtime(createdAt/1000) < ( now() - interval '5' minutes ) or createdAt is null)
order by
    createdAt asc
""")

if (badMeetingDF.count() > 0) {
    badMeetingDF.show(5)
    z.put("BAD_MEETING", true);   
} else {
    z.put("BAD_MEETING", false);
}

Send Out the Email if Needed
import java.util.Date
import javax.mail.Message.RecipientType
import javax.mail.internet.{InternetAddress, MimeMessage}
import javax.mail.{Address, MessagingException, Session, Transport}
import java.util.Properties

val text = "Bad data detected: <br/><br/> " +
           "Here is a link Graph (http://link1) <br/><br/>" +
           "Here is a link Data (http://link2)"

val host = "smtp.gmail.com"
val port = "465"

val recipients = Array( “[e-mail]" )
val username = “[e-mail]"
val password = “xxxxxxx"

val properties = new Properties()
properties.put("mail.smtp.port", port)
properties.put("mail.smtp.auth", "true")
properties.put("mail.smtp.socketFactory.class", "javax.net.ssl.SSLSocketFactory");

val badMeeting = z.get("BAD_MEETING");

if (badMeeting != null) {

    val session = Session.getDefaultInstance(properties, null)
    val message = new MimeMessage(session)
    val addresses: Array[Address] = recipients.map(new InternetAddress(_)).toArray
    message.addRecipients(RecipientType.TO, addresses)
    message.setSubject(subject)
    message.setContent(text, "text/html")
    val transport = session.getTransport("smtp")

    transport.connect(host, username, password)
    transport.sendMessage(message, message.getAllRecipients)
}



References:
https://zeppelin.apache.org/docs/0.8.1/interpreter/shell.html
https://mapr.com/docs/61/Zeppelin/ZeppelinShell.html
https://zeppelin.apache.org/docs/0.8.1/usage/other_features/cron_scheduler.html#enable-cron

Zeppelin and Email Sending
https://zeppelin.apache.org/docs/0.8.1/interpreter/shell.html
https://stackoverflow.com/questions/39225641/automatic-email-alerts-from-zeppelin-notebook
https://coderanch.com/t/679192/java/Sending-email-Java
http://www.tutorialspoint.com/java/java_sending_email.htm
https://www.mkyong.com/java/java-how-to-send-email/
https://www.programcreek.com/scala/javax.mail.internet.MimeMessage

Zeppelin RESTful API
http://zeppelin.apache.org/docs/0.8.1/usage/rest_api/notebook.html#get-the-status-of-all-paragraphs
http://notebook.sillycat.com/zeppelin/api/notebook/2ECXC39A1/paragraph/20190621-012609_607701054
http://notebook.sillycat.com/zeppelin/api/notebook/2ECXC39A1
http://notebook.sillycat.com/zeppelin/api/notebook