NullPointerException в кафке - PullRequest
Новая
       31

NullPointerException в кафке

2 голосов
/ 29 марта 2019

Я хочу записать данные из Kafka в базу данных MySQL. Я реализовал следующий код. 

kafkaConsumer.subscribe(Arrays.asList(topicName));
    try {


        while (true) {
            ConsumerRecords<String, String> record = kafkaConsumer.poll(1000);
            for (ConsumerRecord<String, String> record1 : record) {
                if (record1.value().length() > 0) {
                    System.out.println(record1.value());
                    String value = record1.value();
                    String[] array = value.split(" ");
                    String sql = String.format("insert into data(timestamp, LogLevel,CityName,Detail) values ('%s', '%s','%s','%s')", array[0], array[1], array[2], array[3]);
                    mysqlConnector.statement.executeUpdate(sql);

                }
            }
        }

Ответы [ 2 ]

1 голос
/ 29 марта 2019

Может быть, не стоит изобретать велосипед.Наилучшим вариантом является использование хорошо известного и проверенного инструмента: Kafka Connect с Confluent JDBC Sink Connector

https://docs.confluent.io/current/connect/kafka-connect-jdbc/sink-connector/index.html

0 голосов
/ 29 марта 2019

Пример подключения Java-приложения к базе данных MySQL: в этом примере sonoo - это имя базы данных, root - это имя пользователя и пароль. 

import java.sql.*;  
class MysqlCon{  
public static void main(String args[]){  
try{  
Class.forName("com.mysql.jdbc.Driver");  
Connection con=DriverManager.getConnection(  
"jdbc:mysql://localhost:3306/sonoo","root","root");  
//here sonoo is database name, root is username and password  
Statement stmt=con.createStatement();  
ResultSet rs=stmt.executeQuery("select * from emp");  
while(rs.next())  
System.out.println(rs.getInt(1)+"  "+rs.getString(2)+"  "+rs.getString(3));  
con.close();  
}catch(Exception e){ System.out.println(e);}  
}  
}

Для получения дополнительной информации о подключении базы данных Java с MySQL, проверьте:https://www.javatpoint.com/example-to-connect-to-the-mysql-database

...