flume监控ftp得json文件

src/main/java/com/ihooyah/flume/sink/DAOClass.java

@@ -45,7 +45,7 @@ public class DAOClass {
             String tableName = (String)params.get("tableName");
             String iscustom = (String)params.get("iscustom");
             JSONObject initObject=JSONObject.parseObject(body);
-            JSONArray initArray=JSONArray.parseArray(initObject.get("data").toString());
+            JSONObject dataObject=initObject.getJSONObject("data");
             if (!"false".equals(iscustom)) {
                 if ("true".equals(iscustom)) {
                     ;
@@ -58,7 +58,7 @@ public class DAOClass {
                 ResultSet tableResult = st.executeQuery(isTableExist);
                 if ("false".equals(tableResult.next()+"")) {
                     String createTableSql = "create table flume_json (id int PRIMARY key AUTO_INCREMENT,type varchar(20),data text," +
-                            "file_name varchar(40),create_time datetime);";
+                            "file_name varchar(40),create_time datetime,status varchar(5));";
                     st.execute(createTableSql);
                 }
                 SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@@ -70,15 +70,13 @@ public class DAOClass {
                 DELETE_QUERY="DELETE FROM flume_json where create_time<'"+mon+"';";
                 st.execute(DELETE_QUERY);
                 String values = "";
-                for (Object data : initArray) {
-                    JSONObject dataJson = JSONObject.parseObject(data.toString());
-                    columns="type,data,file_name,create_time";
-                    values="'"+initObject.get("type").toString()+"','"+dataJson+"','"+fileName+"','"+sdf.format(new Date())+"'";
+                columns="type,data,file_name,create_time,status";
+                values="'"+initObject.get("type").toString()+"','"+dataObject+"','"+fileName+"','"+sdf.format(new Date())+"','0'";
                 INSERT_QUERY = "INSERT INTO flume_json (" + columns + ") values (" + values + ")";
                 System.out.println("数据新增语句:"+INSERT_QUERY);
                 PreparedStatement insertStmnt = this.connection.prepareStatement(INSERT_QUERY);
                 insertStmnt.execute();
-                }
+
             }
         } catch (SQLException var20) {
             var20.printStackTrace();