TrumanDu's Blog

List<String>list = Arrays.asList("a","c","b");
Collections.sort(list, (o1,o2)->o1.compareTo(o2));

str->str.toLowerCase()
(o1,o2)->o1.compareTo(o2)
(o1,o2)->{return o1.compareTo(o2)}
(String o1,String o2)->{return o1.compareTo(o2)}

//不指定前缀和后缀
StringJoiner stringJoiner = new StringJoiner(",");
//指定前缀和后缀
//StringJoiner stringJoiner = new StringJoiner(",","{","}");
List<String> list = Arrays.asList("a","b","c");
list.forEach(str->stringJoiner.add(str));

List<String> list = Arrays.asList("a","b","c");
System.out.println(String.join(",", list));

Clock clock = Clock.systemDefaultZone();
long millis = clock.millis();

Instant instant = clock.instant();
Date legacyDate = Date.from(instant);   // legacy java.util.Date

System.out.println(ZoneId.getAvailableZoneIds());
// prints all available timezone ids

ZoneId shanghaiZoneId = ZoneId.of("Asia/Shanghai");
// ZoneRules[currentStandardOffset=+08:00]

LocalDateTime localDateTime = LocalDateTime.now();
ZonedDateTime zonedDateTime = ZonedDateTime.of(localDateTime, shanghaiZoneId);

LocalTime now1 = LocalTime.now(zone1);
LocalTime now2 = LocalTime.now(zone2);

System.out.println(now1.isBefore(now2));  // false

long hoursBetween = ChronoUnit.HOURS.between(now1, now2);
long minutesBetween = ChronoUnit.MINUTES.between(now1, now2);

System.out.println(hoursBetween);       // -3
System.out.println(minutesBetween);     // -239

LocalDate today = LocalDate.now();
LocalDate tomorrow = today.plus(1, ChronoUnit.DAYS);
LocalDate yesterday = tomorrow.minusDays(2);

LocalDate independenceDay = LocalDate.of(2014, Month.JULY, 4);
DayOfWeek dayOfWeek = independenceDay.getDayOfWeek();
System.out.println(dayOfWeek);    // FRIDAY

LocalDateTime sylvester = LocalDateTime.of(2014, Month.DECEMBER, 31, 23, 59, 59);

DayOfWeek dayOfWeek = sylvester.getDayOfWeek();
System.out.println(dayOfWeek);      // WEDNESDAY

Month month = sylvester.getMonth();
System.out.println(month);          // DECEMBER

long minuteOfDay = sylvester.getLong(ChronoField.MINUTE_OF_DAY);
System.out.println(minuteOfDay);    // 1439

ZonedDateTime dateTime = ZonedDateTime.now();
ZonedDateTime zonedDateTime = ZonedDateTime.of(LocalDateTime.now(), shanghaiZoneId);

Instant first = Instant.now();
// wait some time while something happens
Instant second = Instant.now();
Duration duration = Duration.between(first, second);

BASIC_ISO_DATE

ISO_LOCAL_DATE
ISO_LOCAL_TIME
ISO_LOCAL_DATE_TIME

ISO_OFFSET_DATE
ISO_OFFSET_TIME
ISO_OFFSET_DATE_TIME

ISO_ZONED_DATE_TIME

ISO_INSTANT

ISO_DATE
ISO_TIME
ISO_DATE_TIME

ISO_ORDINAL_TIME
ISO_WEEK_DATE

RFC_1123_DATE_TIME

DateTimeFormatter formatter = DateTimeFormatter.ISO_DATE_TIME;
//DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
String formattedDate = formatter.format(LocalDateTime.now());

$ bin/kafka-consumer-groups.sh --describe --bootstrap-server 192.168.0.101:8092 --group test
Consumer group 'test' has no active members.

TOPIC              PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
truman_test_offset 2          1325            2361            1036            -               -               -
truman_test_offset 6          1265            2289            1024            -               -               -
truman_test_offset 4          1245            2243            998             -               -               -
truman_test_offset 9          1310            2307            997             -               -               -
truman_test_offset 1          1259            2257            998             -               -               -
truman_test_offset 8          1410            2438            1028            -               -               -
truman_test_offset 3          1225            2167            942             -               -               -
truman_test_offset 0          1218            2192            974             -               -               -
truman_test_offset 5          1262            2252            990             -               -               -
truman_test_offset 7          1265            2277            1012            -               -               -

public class TopicConsumerGroupState {
	private String groupId;
	/**
	 * 消费方式：zk/broker 主要指的offset提交到哪里 新版本 broker 旧版本zk
	 */
	private String consumerMethod;
	private List<PartitionAssignmentState> partitionAssignmentStates;
	/**
	 * Dead：组内已经没有任何成员的最终状态，组的元数据也已经被coordinator移除了。这种状态响应各种请求都是一个response：
	 * UNKNOWN_MEMBER_ID Empty：组内无成员，但是位移信息还没有过期。这种状态只能响应JoinGroup请求
	 * PreparingRebalance：组准备开启新的rebalance，等待成员加入 AwaitingSync：正在等待leader
	 * consumer将分配方案传给各个成员 Stable：rebalance完成！可以开始消费了~
	 */
	private ConsumerGroupState consumerGroupState;

    //省略set和get方法..

}

public class PartitionAssignmentState {
	private String group; 
	private String topic;
	private int partition;
	private long offset;
	private long lag;
	private String consumerId;
	private String host;
	private String clientId;
	private long logEndOffset;
	//省略set和get方法..
}

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.2.0</version>
</dependency>

/**
	 * 根据topic 获取offset,该结果是不同group 组不同patition的当前消费offset
	 * 
	 * @param topic
	 * @return Map<String, Set<Entry<TopicPartition, OffsetAndMetadata>>>
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TimeoutException
	 */
	public Map<String, Set<Entry<TopicPartition, OffsetAndMetadata>>> listConsumerGroupOffsets(String topic)
			throws InterruptedException, ExecutionException, TimeoutException {
		Set<String> groupIds = this.listConsumerGroups(topic);
		Map<String, Set<Entry<TopicPartition, OffsetAndMetadata>>> consumerGroupOffsets = new HashMap<>();
		groupIds.forEach(groupId -> {
			Set<Entry<TopicPartition, OffsetAndMetadata>> consumerPatitionOffsets = new HashSet<>();
			try {
				consumerPatitionOffsets = this.adminClient.listConsumerGroupOffsets(groupId)
						.partitionsToOffsetAndMetadata().get(30, TimeUnit.SECONDS).entrySet().stream()
						.filter(entry -> topic.equalsIgnoreCase(entry.getKey().topic())).collect(Collectors.toSet());
				;
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			} catch (TimeoutException e) {
				e.printStackTrace();
			}
			consumerGroupOffsets.put(groupId, consumerPatitionOffsets);
		});
		return consumerGroupOffsets;
	}

	/**
	 * 根据topic 获取topicConsumerGroupStates,其中未包含lag/logEndOffset(consumer可见offset)
	 * 
	 * @param topic
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TimeoutException
	 */
	public List<TopicConsumerGroupState> describeConsumerGroups(String topic)
			throws InterruptedException, ExecutionException, TimeoutException {
		final List<TopicConsumerGroupState> topicConsumerGroupStates = new ArrayList<>();
		Set<String> groupIds = this.listConsumerGroups(topic);
		Map<String, ConsumerGroupDescription> groupDetails = this.adminClient.describeConsumerGroups(groupIds).all()
				.get(30, TimeUnit.SECONDS);

		Map<String, Set<Entry<TopicPartition, OffsetAndMetadata>>> consumerPatitionOffsetMap = this
				.listConsumerGroupOffsets(topic);

		groupDetails.entrySet().forEach(entry -> {
			String groupId = entry.getKey();
			ConsumerGroupDescription description = entry.getValue();

			TopicConsumerGroupState topicConsumerGroupState = new TopicConsumerGroupState();
			topicConsumerGroupState.setGroupId(groupId);
			topicConsumerGroupState.setConsumerMethod("broker");
			topicConsumerGroupState.setConsumerGroupState(description.state());
			// 获取group下不同patition消费offset信息
			Set<Entry<TopicPartition, OffsetAndMetadata>> consumerPatitionOffsets = consumerPatitionOffsetMap
					.get(groupId);
			List<PartitionAssignmentState> partitionAssignmentStates = new ArrayList<>();

			if (!description.members().isEmpty()) {
				// 获取存在consumer(memeber存在的情况)
				partitionAssignmentStates = this.withMembers(consumerPatitionOffsets, topic, groupId, description);
			} else {
				// 获取不存在consumer
				partitionAssignmentStates = this.withNoMembers(consumerPatitionOffsets, topic, groupId);
			}
			topicConsumerGroupState.setPartitionAssignmentStates(partitionAssignmentStates);
			topicConsumerGroupStates.add(topicConsumerGroupState);
		});

		return topicConsumerGroupStates;
	}

	private List<PartitionAssignmentState> withMembers(
			Set<Entry<TopicPartition, OffsetAndMetadata>> consumerPatitionOffsets, String topic, String groupId,
			ConsumerGroupDescription description) {
		List<PartitionAssignmentState> partitionAssignmentStates = new ArrayList<>();
		Map<Integer, Long> consumerPatitionOffsetMap = new HashMap<>();
		consumerPatitionOffsets.forEach(entryInfo -> {
			TopicPartition topicPartition = entryInfo.getKey();
			OffsetAndMetadata offsetAndMetadata = entryInfo.getValue();
			consumerPatitionOffsetMap.put(topicPartition.partition(), offsetAndMetadata.offset());
		});
		description.members().forEach(memberDescription -> {
			memberDescription.assignment().topicPartitions().forEach(topicPation -> {
				PartitionAssignmentState partitionAssignmentState = new PartitionAssignmentState();
				partitionAssignmentState.setPartition(topicPation.partition());
				partitionAssignmentState.setTopic(topic);
				partitionAssignmentState.setClientId(memberDescription.clientId());
				partitionAssignmentState.setGroup(groupId);
				partitionAssignmentState.setConsumerId(memberDescription.consumerId());
				partitionAssignmentState.setHost(memberDescription.host());
				partitionAssignmentState.setOffset(consumerPatitionOffsetMap.get(topicPation.partition()));
				partitionAssignmentStates.add(partitionAssignmentState);
			});
		});
		return partitionAssignmentStates;
	}

	private List<PartitionAssignmentState> withNoMembers(
			Set<Entry<TopicPartition, OffsetAndMetadata>> consumerPatitionOffsets, String topic, String groupId) {
		List<PartitionAssignmentState> partitionAssignmentStates = new ArrayList<>();
		consumerPatitionOffsets.forEach(entryInfo -> {
			TopicPartition topicPartition = entryInfo.getKey();
			OffsetAndMetadata offsetAndMetadata = entryInfo.getValue();
			PartitionAssignmentState partitionAssignmentState = new PartitionAssignmentState();
			partitionAssignmentState.setPartition(topicPartition.partition());
			partitionAssignmentState.setTopic(topic);
			partitionAssignmentState.setGroup(groupId);
			partitionAssignmentState.setOffset(offsetAndMetadata.offset());
			partitionAssignmentStates.add(partitionAssignmentState);
		});
		return partitionAssignmentStates;
	}

<dependency>
	<groupId>com.101tec</groupId>
	<artifactId>zkclient</artifactId>
	<version>0.11</version>
</dependency>
<dependency>
	<groupId>org.apache.kafka</groupId>
	<artifactId>kafka_2.11</artifactId>
</dependency>

private List<TopicConsumerGroupState> getZKConsumerOffsets(String topic) {
	final List<TopicConsumerGroupState> topicConsumerGroupStates = new ArrayList<>();
	Set<String> zkGroups = this.listTopicGroups(topic);
	zkGroups.forEach(group -> {
		Map<String, Map<String, String>> zkConsumerOffsets = this.getZKConsumerOffsets(group,
				topic);

		Map<TopicPartition, Long> offsetsMap = this.consumerOffset(group, topic);

		TopicConsumerGroupState topicConsumerGroupState = new TopicConsumerGroupState();
		topicConsumerGroupState.setGroupId(group);
		topicConsumerGroupState.setConsumerMethod("zk");

		List<PartitionAssignmentState> partitionAssignmentStates = new ArrayList<>();

		Map<String, Long> offsetsTempMap = new HashMap<>(offsetsMap.size());
		offsetsMap.forEach((k, v) -> {
			offsetsTempMap.put(k.partition() + "", v);
		});

		zkConsumerOffsets.forEach((patition, topicDesribe) -> {
			Long logEndOffset = offsetsTempMap.get(patition);
			String owner = topicDesribe.get("owner");
			long offset = Long.parseLong(topicDesribe.get("offset"));
			PartitionAssignmentState partitionAssignmentState = new PartitionAssignmentState();
			partitionAssignmentState.setClientId(owner);
			partitionAssignmentState.setGroup(group);
			partitionAssignmentState.setLogEndOffset(logEndOffset);
			partitionAssignmentState.setTopic(topic);
			partitionAssignmentState.setOffset(offset);
			partitionAssignmentState.setPartition(Integer.parseInt(patition));
			partitionAssignmentState.setLag(getLag(offset, logEndOffset));

			partitionAssignmentStates.add(partitionAssignmentState);
		});
		topicConsumerGroupState.setPartitionAssignmentStates(partitionAssignmentStates);
		topicConsumerGroupStates.add(topicConsumerGroupState);
	});
	return topicConsumerGroupStates;
}

npm install antd --save
npm install ant-design-pro@latest --save

// use localStorage to store the authority info, which might be sent from server in actual project.
export function getAuthority() {
  return localStorage.getItem('ice-pro-authority') || 'admin';
}

export function setAuthority(authority) {
  return localStorage.setItem('ice-pro-authority', authority);
}

export function removeAuthority() {
  return localStorage.removeItem('ice-pro-authority');
}

import RenderAuthorized from 'ant-design-pro/lib/Authorized';
import { getAuthority } from './authority';

let Authorized = RenderAuthorized(getAuthority()); // eslint-disable-line

// 更新权限
const reloadAuthorized = () => {
  Authorized = RenderAuthorized(getAuthority());
};

export { reloadAuthorized };
export default Authorized;

this.setState({
      switchStatus: !this.state.switchStatus
    });

<DetailDialog dialogObj={dialogObj} } />

constructor(props) {
      super(props);
      this.state = {
        dialogObj: this.props.dialogObj,
      };
    }

// 接受父类props改变，修改子类中的属性
componentWillReceiveProps(nextProps) {
  this.setState({
    dialogObj: nextProps.dialogObj,
  });
}

<SimpleFormDialog ref="getDialog">

父组件方法
parentMethod = () => {
            //调用组件进行通信
            this.refs.getDialog.childMethod();
}

<DetailDialog hideDetailDialog={this.hideDetailDialog} />

hideDetailDialog = () => {
      this.props.hideDetailDialog();
    }

CREATE STREAM orders (
  orderId BIGINT,
  address STRUCT<street VARCHAR, zip INTEGER>) WITH (...);
  
SELECT address->city, address->zip FROM orders;

sudo docker run -it --rm  confluentinc/cp-ksql-cli:5.1.2  http://192.168.0.11:8088
[sudo] password for bigdata:

                  ===========================================
                  =        _  __ _____  ____  _             =
                  =       | |/ // ____|/ __ \| |            =
                  =       | ' /| (___ | |  | | |            =
                  =       |  <  \___ \| |  | | |            =
                  =       | . \ ____) | |__| | |____        =
                  =       |_|\_\_____/ \___\_\______|       =
                  =                                         =
                  =  Streaming SQL Engine for Apache Kafka® =
                  ===========================================

Copyright 2017-2018 Confluent Inc.

CLI v5.1.2, Server v5.1.2 located at http://192.168.0.11:8088

Having trouble? Type 'help' (case-insensitive) for a rundown of how things work!

ksql> show streams;

 Stream Name      | Kafka Topic              | Format
------------------------------------------------------
 EC_BOT_DETECTION | EC_bot_detection.replica | JSON
 DEMO             | trumantest               | JSON
------------------------------------------------------

curl --request POST \
  --url http://192.168.0.11:8088/ksql \
  --header 'Content-Type: application/vnd.ksql.v1+json; charset=utf-8' \
  --data '{\r\n  "ksql": "LIST STREAMS;",\r\n  "streamsProperties": {}\r\n}'

CREATE STREAM ec_bot_detection (action VARCHAR, nvtc VARCHAR,nid VARCHAR,ip VARCHAR,msg VARCHAR,level VARCHAR,verb VARCHAR,url VARCHAR,purl VARCHAR,ua VARCHAR,valid INTEGER,nNvtc VARCHAR,nNid VARCHAR,cip VARCHAR,sip VARCHAR,xpost VARCHAR,time VARCHAR,id VARCHAR,xprotocol VARCHAR,xhost VARCHAR,xport VARCHAR,xpath VARCHAR,xquery VARCHAR,xpathalias VARCHAR,xkey VARCHAR,importance INTEGER,xpprotocol VARCHAR,xphost VARCHAR,xpport VARCHAR,xppath VARCHAR,xpquery VARCHAR,org VARCHAR,cyc VARCHAR,isp VARCHAR,cycName VARCHAR,region VARCHAR,city VARCHAR,geoLocation VARCHAR,zipcode VARCHAR) WITH (VALUE_FORMAT = 'JSON',KAFKA_TOPIC = 'EC_bot_detection.replica');

SELECT ip,org,cyc,count(1) as ccount FROM EC_BOT_DETECTION WINDOW TUMBLING (SIZE 60 SECONDS) WHERE xhost not like '%.tw' and level='-1' AND importance=1 AND (msg<>'W' OR msg is null) and action like'crawler%' AND ip NOT LIKE '127.0.0.1' AND ip NOT LIKE '172.16.%' AND ip <> ''  AND xpathalias !='LandingpageOverviewcontent4mobile'  AND xpathalias !='CommonCommonrecaptchavalidate'  GROUP BY ip,org,cyc HAVING count(1)>300;

CREATE TABLE bot_detection_1min as SELECT ip,org,cyc,count(1) as ccount FROM EC_BOT_DETECTION WINDOW TUMBLING (SIZE 60 SECONDS) WHERE xhost not like '%.tw' and level='-1' AND importance=1 AND (msg<>'W' OR msg is null) and action like'crawler%' AND ip NOT LIKE '127.0.0.1' AND ip NOT LIKE '172.16.%' AND ip <> ''  AND xpathalias !='LandingpageOverviewcontent4mobile'  AND xpathalias !='CommonCommonrecaptchavalidate'  GROUP BY ip,org,cyc HAVING count(1)>300;

@UdfDescription(name = "upper", description = "字符串转大写")
public class Upper {
	@Udf
	public String upper(final String column) {
		return column.toUpperCase();
	}

}

@UdafDescription(name = "percent", description = "聚合求百分比")
public class Percent {
	@UdafFactory(description = "statistics percent")
	public static Udaf<Integer, Map<String, Double>> createStddev() {
		return new Udaf<Integer, Map<String, Double>>() {
			@Override
			public Map<String, Double> initialize() {
				Map<String, Double> stats = new HashMap<>();
				stats.put("sum", 0.0);
				stats.put("bussiness", 0.0);
				stats.put("percent", 0.0);
				return stats;

			}

			@Override
			public Map<String, Double> aggregate(final Integer val, final Map<String, Double> aggregateValue) {
				Double sum = aggregateValue.getOrDefault("sum", 0.0) + 1;
				Double bussiness = aggregateValue.getOrDefault("bussiness", 0.0);
				if (val % 4 == 0) {
					bussiness = bussiness + 1;
				}

				Double percent = bussiness / sum;

				Map<String, Double> agg = new HashMap<>();
				agg.put("sum", sum);
				agg.put("bussiness", bussiness);
				agg.put("percent", percent);
				return agg;
			}

			@Override
			public Map<String, Double> merge(final Map<String, Double> aggOne, final Map<String, Double> aggTwo) {

				Double sum = aggOne.getOrDefault("sum", 0.0) + aggTwo.getOrDefault("sum", 0.0);
				Double bussiness = aggOne.getOrDefault("bussiness", 0.0) + aggTwo.getOrDefault("bussiness", 0.0);
				Double percent = bussiness / sum;

				Map<String, Double> agg = new HashMap<>();
				agg.put("sum", sum);
				agg.put("bussiness", bussiness);
				agg.put("percent", percent);
				return agg;
			}
		};

	}
}

$ pip install PyGithub

from github import Github
import os
import codecs
def create(newfile):
    if not os.path.exists(newfile):
         f = open(newfile,'w')
         print newfile
         f.close()
    return
fileName='rct.txt'
respository='xaecbd/RCT'
g=Github("token",timeout=300)

repo=g.get_repo(respository)
stargazers=repo.get_stargazers_with_dates()
create(fileName)
with codecs.open(fileName,'a', 'utf-8') as f:
    for people in stargazers:
        print people.user.email
        if people.user.email is not None:
            if people.user.name is None:
                name = people.user.email
            else:
                name = people.user.name
            data=(name)+":"+(people.user.email)+"\r\n"
            f.write(data)

python crawer.py