发表于 | | 阅读次数:

先看一个简单的KafkaConsumer例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
Properties props = new Properties();  
props.put("bootstrap.servers", "localhost:9092");  
props.put("group.id", "test_group");  
props.put("enable.auto.commit", "true");  
props.put("auto.commit.interval.ms", "1000");  
props.put("session.timeout.ms", "30000");  
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);  
consumer.subscribe(Arrays.asList("test_topic"));  
while (true) {  
    ConsumerRecords<String, String> records = consumer.poll(100);  
    for (ConsumerRecord<String, String> record : records){  
        System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());  
    }  
}
阅读全文 »

发表于 | | 阅读次数:

Kafka0.9发布了新版consumer client。它与旧版本最大的区别是移除了基于zookeeper的high-level consumer和low-level SimpleConsumer，而代之于一个统一的consumer API，它集成了之前high-level consumer的group管理功能和low-level consumer的offset控制功能。

阅读全文 »

发表于 | | 阅读次数:

Kafka Connect是Kafka0.9新增的模块。可以从名称看出，它可以和外部系统、数据集建立一个数据流的连接，实现数据的输入、输出。有以下特性：

  • 使用了一个通用的框架，可以在这个框架上非常方面的开发、管理Kafka Connect接口
  • 支持分布式模式或单机模式进行运行
  • 支持REST接口，可以通过REST API提交、管理 Kafka Connect集群
  • offset自动管理
阅读全文 »

发表于 | | 阅读次数:
  • 原生ES，只需要在启动时加上-Xmx1g -Xms1g
  • Elasticsearch Docker，参数为 -eES_MIN_MEM=1g-eES_MAX_MEM=1g

注意：

  1. mx和ms最好设为一样，避免GC
  2. 为了保证最大效率，ES内存设为预留内存的一半，另外一半留给Lucene。

发表于 | | 阅读次数:

项目中碰到一个elasticsearch的purge需求，就是说在不删除index和type的前提下，清除其中的所有数据。

用es的delete by query api可以做到，尽管官方声明deprecated in 1.5.3，但是经过测试1.8还是可以用的（真不敢想象要是不能用了怎么办。。。连个purge的api都没有。。），语法如下：

阅读全文 »

发表于 | | 阅读次数:

Elasticsearch Bulk API允许批量提交index和delete请求。

1
2
3
4
BulkRequestBuilder bulkRequest = client.prepareBulk();  
bulkRequest.add(client.prepareIndex("index1", "type1", "id1").setSource(source);  
bulkRequest.add(client.prepareIndex("index2", "type2", "id2").setSource(source);  
BulkResponse bulkResponse = bulkRequest.execute().actionGet();

但有时我们需要更精细的批量操控，比如

阅读全文 »

发表于 | | 阅读次数:

最近尝试用Proguard来混淆代码，以增加发布出去的代码安全性。今天运行混淆后的jar包发生如下异常

1
2
3
4
5
Caused by: org.springframework.beans.factory.BeanCreationException: Could not autowire field: public abc.service.Service test.rest.controller.Controller.service; nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type [abc.service.Service] found for dependency: expected at least 1 bean which qualifies as autowire candidate for this dependency. Dependency annotations: {@org.springframework.beans.factory.annotation.Autowired(required=true)}
at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor$AutowiredFieldElement.inject(AutowiredAnnotationBeanPostProcessor.java:561)
at org.springframework.beans.factory.annotation.InjectionMetadata.inject(InjectionMetadata.java:88)
at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor.postProcessPropertyValues(AutowiredAnnotationBeanPostProcessor.java:331)
... 26 more
阅读全文 »

发表于 | | 阅读次数:

最近在做一个spark项目，顺便分享一下我的Scala入门过程。这一系列文章假定读者有一定的java或者其他面向对象编程语言基础。本文主要简单介绍文件的操作。

按惯例先上代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
val file = Source.fromFile("/Users/xiejing/Desktop/javascript")  
for (line <- file.getLines()) {  
  println(line)  
}  
file.close()  
  
val webFile = Source.fromURL("https://www.baidu.com/");  
webFile.foreach(print)  
webFile.close  
  
var javaWriter = new BufferedWriter(new FileWriter(new File("/Users/xiejing/Desktop/a.txt")))  
for (i <- 4 to 10) {  
  javaWriter.write(i.toString())  
}  
javaWriter.flush()  
javaWriter.close()
阅读全文 »

发表于 | | 阅读次数:

最近在做一个spark项目，顺便分享一下我的Scala入门过程。这一系列文章假定读者有一定的java或者其他面向对象编程语言基础。本文主要简单介绍Scala的集合类型，包括Array，Tuple和Map。

照例先上一段代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
object TestScala3 {  
  def main(args: Array[String]): Unit = {  
    val tuple = (1, "two", "three", true)  
    println(tuple._1)  
    println(tuple._2)  
  
    val array = Array(1, "two", "three", true)  
    for (i <- 0 until array.length) {  
      println(array(i))  
    }  
    for (item <- array) println(item)  
  
    val map = Map(1 -> "one", "two" -> 2, true -> "true")  
    for ((k, v) <- map) println("key = " + k + ", value = " + v)  
    for ((_, v) <- map) println("value = " + v)  
    println(map.get(1))  
    println(map.get(3))  
  }  
}
阅读全文 »