Druid.io: update/override existing data via streams from Kafka (Druid Kafka indexing service)

2018-03-17 14:32 Bo. imported from Stackoverflow

I'm loading streams from Kafka using the Druid Kafka indexing service.

But the data I uploaded is always changed, so I need to reload it again and avoid duplicates and collisions if data was already loaded.

I research docs about Updating Existing Data in Druid.

But all info about Hadoop Batch Ingestion, Lookups .

Is it possible to update existing Druid data during Kafka streams?

In other words, I need to rewrite the old values with new ones using Kafka indexing service (streams from Kafka).

May be any kind of setting to rewrite duplicates?

See also questions close to this topic

is it possible to encrypt 10 digit number to 4 digit number using any algorithm?

Example 10 digit number 9898989898 this should be encrypt to 4 or 6 digit number after that if i decrypt that 4 or 6 digit number its same as 9898989898. Is their any algorithm to do this please help me on this
What is best way for syncing files from local computer to Android mobile and vice versa when no internet connection?

I have developed a project in which i am doing import and export functionality using usb. The scenario is that local machine send me some commands and i read that commmand and accordingly folders are created in mobile. but sometimes it crashes and also according to observation among 100 mobiles it works in 97 but stop in 3 mobiles and i am not getting exact error. so is there any other way for syncing when both are offline

javamail cant send email as a standalone jar

I have created a program in java where i create some files and then i am creating a multipart email message and send it to the coresponded recipients. My problem is that when i execute the program from inside the netbeans it works all fine. But i want to automate it to do it for everyday so i want to use it as a standalone jar. When i am using the jar it works all well until it creates the files, but when it comes at the the point to create the email and send it it does nothing. I have debugged it with print messages but it shows no error messages, it justs stops before it creates the email.

Any suggestions?

try {
            sfpath2 = path2 + "\\" + syear + "-" + smonth + "\\";
            Files.createDirectories(Paths.get(sfpath2));
            new ExportToExcel().createWorkbook(sfpath2, io, "NARACC", rtudata1, rtudata2, rtudata11, rtudata21, rtuave21);
            new ExportToExcel().createWorkbook1(sfpath2, io, "SARACC", rtudata1, rtudata2, rtudata11, rtudata21, rtuave11);
            int jj = io.indexOf("P");
            String ff1 = "NARACC " + io.substring(3, jj - 1).replaceAll("_", "-") + ".xls";
            writet(date);
            writet(sfpath2);
            writet(ff1);
            MailSend mail = new MailSend(date, sfpath2, ff1);
            mail.SendTest("Test");
        } catch (Exception e) {
            e.printStackTrace();
        }

is this ok?

"io" is name of a file. writet() is a method to write the messages to a file.

Here is the code of the mail:

public class MailSend {

String SMTP_HOST_NAME = "";
String SMTP_AUTH_USER = ";
String SMTP_AUTH_PWD = "";
String Date = "";
String Path = "";
String file = "";

public MailSend(String Date, String path, String f) {
    JumpHosts.writet(this.Path + " " + this.file + " 1");
    this.SMTP_HOST_NAME = "";
    JumpHosts.writet(this.Path + " " + this.file + " 2");
    this.SMTP_AUTH_USER = "";
    JumpHosts.writet(this.Path + " " + this.file + " 3");
    this.SMTP_AUTH_PWD = "";
    JumpHosts.writet(this.Path + " " + this.file + " 4");
    this.Date = Date;
    JumpHosts.writet(this.Path + " " + this.file + " 5");
    this.Path = path;
    JumpHosts.writet(this.Path + " " + this.file + " 6");
    this.file = f;
    JumpHosts.writet(this.Path + " " + this.file + " 7");
}

public void SendTest(String emailMsgTxt)
        throws Exception {
    String emailSubjectTxt = "";
    emailSubjectTxt = "NARACC " + this.Date;
    String emailFromAddress = "";

    String[] emailListTo = {""};
    String[] emailListCC = {""};
    String[] emailListBCC = {""};

    JumpHosts.writet(this.Path + " " + this.file);
    MailSend smtpMailSender = new MailSend(this.Date, this.Path, this.file);
    JumpHosts.writet(smtpMailSender.Path + " " + smtpMailSender.file);
    smtpMailSender.postMail(emailListTo, emailListCC, emailListBCC, emailSubjectTxt, emailMsgTxt, emailFromAddress);
    JumpHosts.writet("\nSucessfully Sent mail to All Users");
    System.out.println("\nSucessfully Sent mail to All Users");
}

public void postMail(String[] recipients, String[] recipientsCC, String[] recipientsBCC, String subject, String message, String from)
        throws MessagingException {
    boolean debug = false;

    Properties props = new Properties();
    props.put("mail.smtp.host", "");
    props.put("mail.smtp.socketFactory.port", "");
    props.put("mail.smtp.socketFactory.class", "javax.net.ssl.SSLSocketFactory");

    props.put("mail.smtp.auth", "true");
    props.put("mail.smtp.port", "");

    Authenticator auth = new SMTPAuthenticator();
    Session session = Session.getDefaultInstance(props, auth);

    session.setDebug(debug);
    try {
        Message msg = new MimeMessage(session);

        InternetAddress addressFrom = new InternetAddress(from);
        msg.setFrom(addressFrom);

        InternetAddress[] addressTo = new InternetAddress[recipients.length];
        for (int i = 0; i < recipients.length; i++) {
            addressTo[i] = new InternetAddress(recipients[i]);
        }
        msg.setRecipients(Message.RecipientType.TO, addressTo);

        InternetAddress[] addressCC = new InternetAddress[recipientsCC.length];
        for (int i = 0; i < recipientsCC.length; i++) {
            addressCC[i] = new InternetAddress(recipientsCC[i]);
        }
        msg.setRecipients(Message.RecipientType.CC, addressCC);

        InternetAddress[] addressBCC = new InternetAddress[recipientsBCC.length];
        for (int i = 0; i < recipientsBCC.length; i++) {
            addressBCC[i] = new InternetAddress(recipientsBCC[i]);
        }
        msg.setRecipients(Message.RecipientType.BCC, addressBCC);

        msg.setSubject(subject);
        msg.setSentDate(new Date());

        Multipart multipart = new MimeMultipart();

        BodyPart attachmentBodyPart = new MimeBodyPart();
        DataSource source = new FileDataSource(this.Path + this.file);
        attachmentBodyPart.setDataHandler(new DataHandler(source));
        attachmentBodyPart.setFileName(this.file);
        multipart.addBodyPart(attachmentBodyPart);

        BodyPart messageBodyPart = new MimeBodyPart();
        String ss = "<html><body><p style=\"font-family:'Times New Roman'\" style=\"color:black\"><br/>";
        ss += "";            
        messageBodyPart.setContent(ss, "text/html;charset=\"windows-1253\"");
        multipart.addBodyPart(messageBodyPart);
        msg.setContent(multipart);

        msg.saveChanges();
        Transport.send(msg);
    } catch (MessagingException e) {
        JumpHosts.writet(e.toString());
        throw new RuntimeException(e);            
    }
}

private class SMTPAuthenticator
        extends Authenticator {

    private SMTPAuthenticator() {
    }

    public PasswordAuthentication getPasswordAuthentication() {
        String username = "";
        String password = "";
        return new PasswordAuthentication(username, password);
    }
}

}

The strings are empty due to obvious reasons. In the real program they are filled with the real data

Python Kafka Streaming API - Binning

I am using python kafka stream binning example given in this, Python Kafka Streaming API

I am able to generate the data using generator.py file given under winton-kafka-streams/examples/binning/, whereas when i run the binning.py file from the same folder, i got the below issue. Could someone help me, to resolve this?
Unknown magic byte with kafka-avro-console-consumer
I have been trying to connect with kafka-avro-console-consumer from Confluent to our legacy Kafka cluster, which was deployed without Confluent Schema Registry. I provided schema explicitly using properties like:
```
kafka-console-consumer --bootstrap-server kafka02.internal:9092 \
    --topic test \
    --from-beginning \
    --property key.schema='{"type":"long"}' \
    --property value.schema='{"type":"long"}'
```
but I am getting 'Unknown magic byte!' error with org.apache.kafka.common.errors.SerializationException

Is it possible to consume Avro messages from Kafka using Confluent kafka-avro-console-consumer that were not serialized with AvroSerializer from Confluent and with Schema Registry?
generate json with backslash using jackson
.
I am using Jackson library to generate json. Jackson is generating the json in below format.

{"color":"yellow","type":"renault"}

I want to generate the above format with backslash as below:

{\"color\":\"yellow\",\"type\":\"renault\"}.

The format with the backslash is required as I am doing transformation using kafka stream API.

Here is my model Class:

public class Car {
```
private String color;
private String type;

public Car() {
}

public Car(final String color, final String type) {
    this.color = color;
    this.type = type;
}

public String getColor() {
    return color;
}

public void setColor(final String color) {
    this.color = color;
}

public String getType() {
    return type;
}

public void setType(final String type) {
    this.type = type;
}
```
}

Please do let me know what configuration/property change do I need to do in order to generate the json in the above required format.

I have also attached the screenshot to show rest of my code.

Thanks in advance!

code screenshot
Intercommunication between Vespa application

I want to create an application in Vespa whcih fetch the data from some appications and feed into another applications.Example:

I have three applications which have data named Location , User and Banner.Now what I want to create an app in vespa which fetch the data from Locaton and User and fed to Banner.

How can I do that? Please help.
I am new to MongoDB, I am creating Movie Record application, here I want to use Indexes but I don't know where can I use them and why?

This is the first time I am using Indexes, I am confused why we need them and how to implement them into our query.

locating values in (a very large) pandas dataframe and store to dictionaries

I have a very large pandas dataframe. The dataframe looks like this:

>> df
    "a_1"   "a_2"   "b_1"  "c_2"  ...
"d_1" nan   0.2   nan  nan
"d_2" 0.1   nan   nan   1
"e_1" nan   1     nan  0.2
"e_2" nan   0.05  0.1  0.7
"f_2" 0.2   0.5   0.3  0.9
...

Now I am trying to go through a list of tuples which contains some row and col names:

t = [("d", "a"), ("d", "c") ...]

For example, when i = ("d", "a"), I want to find out the values corresponded to a_1 and a_2, d_1 and d_2 and I used the following code to locate the values:

s = df.loc[["d_1", "d_2" ], ["a_1", "a_2"]]

# print(s)
#       "a_1"  "a_2"
# "d_1"  nan    0.2
# "d_2   0.1    nan

# convert to list and sort the values
s = s.unstack().reset_index()
s.columns = ["A","B", "Score"]
scores = s.sort_values(by="Score", ascending=False).reset_index(drop=True)

# pick the index(rank) I want and save the not-nan data to dictionary 
rank = 1
try:
    s = scores.loc[rank,:]
except Exception:
    continue

if str(s.Score) != "nan":
    d[(s.A, s.B)] = s.Score # output dictionary

Now the above code works but takes too long, given len(t) = 28350 and I need to test 150+ sets of parameters. One iteration (one set of parameter) takes 3.5mins on the cluster.

I am wondering if there is a better solution to this problem. Thanks in advance!

Pentaho Kettle - Get the file names dynamically

I hope this message finds everyone well!

I'm stucked on a situation on Pentaho PDI Tool and I'm looking for an answer (or at least a light in the end of the cave) to solve it!

I have to import, every month, a bunch of xls's files of differents clients. Every file has a different name (witch is given aleatory) and this files are on a folder named with the name of the client. However, I use the same process for all clients and situations.

Is there a way to pass the name of the directory as a variable and change this variable on every process? How can I read this files on differents paths?
Getting "Execute Process Task Error: File/Process > "*C:\Users\username\pythonProgram.exe*" not in path" while running via SSISDB

I am trying to automate a SSIS package which consists of a Execute Process Task and a Execute SQL task. I am calling a python executable via Process Task

While running the package in SSDT(Visual Studio), it runs perfectly fine but when i try to deploy it in Management Studio(SSISDB), execute process task throws up following error.

"Execute Process Task Error: File/Process "C:\Users\username\pythonProgram.exe" not in path".

While trying to debug, i removed the execute process task in the same package, deployed a fresh copy of the package and it ran successfully even via Management Studio. I am sure it is the python exe file path causing problem.

I tried to run via DTexec.exe cmd utility and facing same problems(running successfully via SSDT but not via Management Studio). Here is the command i used "C:\Program Files (x86)\Microsoft SQL Server\140\DTS\Binn\DTExec.exe" /F "C:\Users\Username\source\repos\TaxAutomation\Package.dtsx" >log.txt"

After seeing various articles, i asked my DBA to put my .exe file in the Server where Management Studio is located and facing same error.

I am trying to resolve this for last two days and worried that me whole work could be undone if i am not able to automate the package by deploying it in SSISDB or using dtexe cmd utlity. My lack of DBA expertise is making me frustrated and anxious.

I will really appreciate help on this.

Generating a single output file for each processed input file in Apach Flink

I am using Scala and Apache Flink to build an ETL that reads all the files under a directory in my local file periodically system and write the result of processing each file in a single output file under another directory.

So an example of this is would be:

/dir/to/input/files/file1
/dir/to/intput/files/fil2
/dir/to/input/files/file3

and the output of the ETL would be exactly:

/dir/to/output/files/file1
/dir/to/output/files/file2
/dir/to/output/files/file3

I have tried various approaches including reducing the parallel processing to one when writing to the dataSink but I still can't achieve the required result.

This is my current code:

   val path = "/path/to/input/files/"
   val format = new TextInputFormat(new Path(path))
   val socketStream = env.readFile(format, path, FileProcessingMode.PROCESS_CONTINUOUSLY, 10)


   val wordsStream = socketStream.flatMap(value => value.split(",")).map(value => WordWithCount(value,1))

   val keyValuePair = wordsStream.keyBy(_.word)

   val countPair = keyValuePair.sum("count")

   countPair.print()

   countPair.writeAsText("/path/to/output/directory/"+
     DateTime.now().getHourOfDay.toString
     +
     DateTime.now().getMinuteOfHour.toString
     +
     DateTime.now().getSecondOfMinute.toString
     , FileSystem.WriteMode.NO_OVERWRITE)

// The first write method I trid:

   val sink = new BucketingSink[WordWithCount]("/path/to/output/directory/")
   sink.setBucketer(new DateTimeBucketer[WordWithCount]("yyyy-MM-dd--HHmm"))

// The second write method I trid:

   val sink3 = new BucketingSink[WordWithCount]("/path/to/output/directory/")
   sink3.setUseTruncate(false)
   sink3.setBucketer(new DateTimeBucketer("yyyy-MM-dd--HHmm"))
   sink3.setWriter(new StringWriter[WordWithCount])
   sink3.setBatchSize(3)
   sink3.setPendingPrefix("file-")
   sink3.setPendingSuffix(".txt")

Both writing methods fail in producing the wanted result.

Can some with experience with Apache Flink guide me to the write approach please.

Calculate 90th percentile for a particular metrics in DRUID

Let say, we have below metrics:

Inbound_Value Outbound_Value,

getting AVG and MAX are easy from DRUID, but how to get 90th Percentile for Inbound_Value or Outbound_Value.

Count Distinct of a dimension in druid

I am trying to get distinct of the count of a dimension territory. But I think what I am getting is distinct of the count of (date, territory). Here is the query:

{
   "queryType": "groupBy",
   "dataSource": "model",
   "granularity": "all",
   "filter": {
   "type": "selector", "dimension": "District", "value": "ATLANTA"},


   "aggregations":[  
      {"name":"!t5", "type":"distinctCount","fieldName":"Territory"},
      {"name":"!T_4","type":"longSum","fieldName":"Units"}
   ],
   "postAggregations":[  
      {  
         "type":"arithmetic",
         "fn":"*",
         "fields":[  
            {  
               "type":"arithmetic",
               "fn":"/",
               "fields":[  
                  {  
                     "type":"fieldAccess",
                     "fieldName":"!T_4"
                  },
                  {  
                     "type":"fieldAccess",
                     "fieldName":"!t5"
                  }
               ]
            },
            {  
               "type":"constant",
               "value":1
            }
         ],
         "name":"AverageDistrict"
      }
   ],
   "intervals": ["2018-06-13T00:00Z/2018-07-25T00:00Z"]
}

so when I am changing the date range, the count of distinct(!t5) also increases. Whereas in the raw data the count of distinct is always constant throughout all the date ranges. What I need is the count of distinct values of territory for that date range. I hope I have made myself clear.

Parquet File upload in Druid

I am new to the Druid. I have done a local setup of druid in local and I am able to load files of json data in druid.

However when I am trying to upload a parquet file, it is giving Unexpected character exception.

I've installed Parquet and Avro extensions and I get following error in each case

$ >curl -X 'POST' -H 'Content-Type:application/json' -d @examples/wikipedia_hadoop_parquet_job.json http://localhost:8090/druid/indexer/v1/task

<html>
<head>
<meta http-equiv="Content-Type" content="text/html;charset=ISO-8859-1"/>
<title>Error 500 </title>
</head>enter code here
<body>
<h2>HTTP ERROR: 500</h2>
<p>Problem accessing /druid/indexer/v1/task. Reason:
<pre>    javax.servlet.ServletException: com.fasterxml.jackson.core.JsonParseException: Unexpected character (&apos;}&apos; (code 125)): was expecting double-quote to start field name
 at [Source: HttpInputOverHTTP@57df7615[c=2032,q=1,[0]=EOF,s=STREAM]; line: 1, column: 493]</pre></p>
<hr /><a href="http://eclipse.org/jetty">Powered by Jetty:// 9.3.19.v20170502</a><hr/>
</body>
</html>

Below is the JSON config file

{
  "type": "index_hadoop",
  "spec": {
    "ioConfig": {
      "type": "hadoop",
      "inputSpec": {
        "type": "static",
        "inputFormat": "org.apache.druid.data.input.parquet.DruidParquetInputFormat",
        "paths": "example/wikipedia_list.parquet"
      },
      "metadataUpdateSpec": {
        "type": "postgresql",
        "connectURI": "jdbc:postgresql://localhost/druid",
        "user" : "druid",
        "password" : "asdf",
        "segmentTable": "druid_segments"
      },

    },
    "dataSchema": {
      "dataSource": "wikipedia",
      "parser": {
        "type": "parquet",
        "parseSpec": {
          "format": "timeAndDims",
          "timestampSpec": {
            "column": "timestamp",
            "format": "auto"
          },
          "dimensionsSpec": {
            "dimensions": [
              "page",
              "language",
              "user",
              "unpatrolled"
            ],
            "dimensionExclusions": [],
            "spatialDimensions": []
          }
        }
      },
      "metricsSpec": [{
        "type": "count",
        "name": "count"
      }, {
        "type": "doubleSum",
        "name": "deleted",
        "fieldName": "deleted"
      }, {
        "type": "doubleSum",
        "name": "delta",
        "fieldName": "delta"
      }],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "DAY",
        "queryGranularity": "NONE",
        "intervals": ["2013-08-30/2013-09-02"]
      }
    },
    "tuningConfig": {
      "type": "hadoop",
      "workingPath": "tmp/working_path",
      "partitionsSpec": {
        "targetPartitionSize": 5000000
      },
      "leaveIntermediate": true
    }
  }
}

I am not able to figure out the issue. Let me know, if I am missing something.

Related questions

Druid.io: update/override existing data via streams from Kafka (Druid Kafka indexing service)

See also questions close to this topic