This is an automated email from the ASF dual-hosted git repository. sjwiesman pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 57267f277339f7ba3790c8e0110cdb77e9593073 Author: Seth Wiesman <[email protected]> AuthorDate: Wed Jun 10 16:06:13 2020 -0500 [FLINK-18194][walkthroughs] Remove table walkthrough archetype --- flink-end-to-end-tests/run-nightly-tests.sh | 2 - .../test-scripts/test_table_walkthroughs.sh | 76 ----------------- .../table/BoundedTransactionTableSource.java | 55 ------------ .../common/table/SpendReportTableSink.java | 99 ---------------------- .../common/table/TruncateDateToHour.java | 46 ---------- .../table/UnboundedTransactionTableSource.java | 89 ------------------- 6 files changed, 367 deletions(-) diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh index 21ea06e..31cf2cd 100755 --- a/flink-end-to-end-tests/run-nightly-tests.sh +++ b/flink-end-to-end-tests/run-nightly-tests.sh @@ -189,8 +189,6 @@ run_test "Elasticsearch (v6.3.1) sink end-to-end test" "$END_TO_END_DIR/test-scr run_test "Quickstarts Java nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_quickstarts.sh java" run_test "Quickstarts Scala nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_quickstarts.sh scala" -run_test "Walkthrough Table Java nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_table_walkthroughs.sh java" -run_test "Walkthrough Table Scala nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_table_walkthroughs.sh scala" run_test "Walkthrough DataStream Java nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_datastream_walkthroughs.sh java" run_test "Walkthrough DataStream Scala nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_datastream_walkthroughs.sh scala" diff --git a/flink-end-to-end-tests/test-scripts/test_table_walkthroughs.sh b/flink-end-to-end-tests/test-scripts/test_table_walkthroughs.sh deleted file mode 100755 index b8f4db5..0000000 --- a/flink-end-to-end-tests/test-scripts/test_table_walkthroughs.sh +++ /dev/null @@ -1,76 +0,0 @@ -#!/usr/bin/env bash -################################################################################ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -################################################################################ - -# End to end test for quick starts test. -# Usage: -# FLINK_DIR=<flink dir> flink-end-to-end-tests/test-scripts/test_table_walkthroughs.sh <Type (java or scala)> - -source "$(dirname "$0")"/common.sh - -TEST_TYPE=$1 - -mkdir -p "${TEST_DATA_DIR}" -cd "${TEST_DATA_DIR}" - -ARTIFACT_ID=flink-walkthrough-table-${TEST_TYPE} -ARTIFACT_VERSION=0.1 - -run_mvn archetype:generate \ - -DarchetypeGroupId=org.apache.flink \ - -DarchetypeArtifactId=flink-walkthrough-table-${TEST_TYPE} \ - -DarchetypeVersion=${FLINK_VERSION} \ - -DarchetypeCatalog=local \ - -DgroupId=org.apache.flink.walkthrough \ - -DartifactId=${ARTIFACT_ID} \ - -Dversion=${ARTIFACT_VERSION} \ - -Dpackage=org.apache.flink.walkthrough \ - -DinteractiveMode=false - -cd "${ARTIFACT_ID}" - -run_mvn clean package > compile-output.txt - -if [[ `grep -c "BUILD FAILURE" compile-output.txt` -eq '1' ]]; then - echo "Failure: The walk-through did not successfully compile" - cat compile-output.txt - exit 1 -fi - -cd target -jar tvf ${ARTIFACT_ID}-${ARTIFACT_VERSION}.jar > contentsInJar.txt - -if [[ `grep -c "org/apache/flink/api/java" contentsInJar.txt` -eq '0' && \ - `grep -c "org/apache/flink/streaming/api" contentsInJar.txt` -eq '0' && \ - `grep -c "org/apache/flink/streaming/experimental" contentsInJar.txt` -eq '0' && \ - `grep -c "org/apache/flink/streaming/runtime" contentsInJar.txt` -eq '0' && \ - `grep -c "org/apache/flink/streaming/util" contentsInJar.txt` -eq '0' ]]; then - - echo "Success: There are no flink core classes are contained in the jar." -else - echo "Failure: There are flink core classes are contained in the jar." - exit 1 -fi - -TEST_PROGRAM_JAR=${TEST_DATA_DIR}/${ARTIFACT_ID}/target/${ARTIFACT_ID}-${ARTIFACT_VERSION}.jar - -add_optional_lib "table" - -start_cluster - -${FLINK_DIR}/bin/flink run "$TEST_PROGRAM_JAR" diff --git a/flink-walkthroughs/flink-walkthrough-common/src/main/java/org/apache/flink/walkthrough/common/table/BoundedTransactionTableSource.java b/flink-walkthroughs/flink-walkthrough-common/src/main/java/org/apache/flink/walkthrough/common/table/BoundedTransactionTableSource.java deleted file mode 100644 index e51a341..0000000 --- a/flink-walkthroughs/flink-walkthrough-common/src/main/java/org/apache/flink/walkthrough/common/table/BoundedTransactionTableSource.java +++ /dev/null @@ -1,55 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.flink.walkthrough.common.table; - -import org.apache.flink.annotation.PublicEvolving; -import org.apache.flink.api.common.io.InputFormat; -import org.apache.flink.api.common.typeinfo.Types; -import org.apache.flink.table.api.TableSchema; -import org.apache.flink.table.sources.InputFormatTableSource; -import org.apache.flink.table.types.DataType; -import org.apache.flink.types.Row; -import org.apache.flink.walkthrough.common.source.TransactionRowInputFormat; - -/** - * A table source for reading a bounded set of transaction events. - * - * <p>This could be backed by a table, database, or other static data set. - */ -@PublicEvolving -@SuppressWarnings({"deprecation", "unused"}) -public class BoundedTransactionTableSource extends InputFormatTableSource<Row> { - @Override - public InputFormat<Row, ?> getInputFormat() { - return new TransactionRowInputFormat(); - } - - @Override - public DataType getProducedDataType() { - return getTableSchema().toRowDataType(); - } - - @Override - public TableSchema getTableSchema() { - return TableSchema.builder() - .field("accountId", Types.LONG) - .field("timestamp", Types.SQL_TIMESTAMP) - .field("amount", Types.DOUBLE) - .build(); - } -} diff --git a/flink-walkthroughs/flink-walkthrough-common/src/main/java/org/apache/flink/walkthrough/common/table/SpendReportTableSink.java b/flink-walkthroughs/flink-walkthrough-common/src/main/java/org/apache/flink/walkthrough/common/table/SpendReportTableSink.java deleted file mode 100644 index 5073d90..0000000 --- a/flink-walkthroughs/flink-walkthrough-common/src/main/java/org/apache/flink/walkthrough/common/table/SpendReportTableSink.java +++ /dev/null @@ -1,99 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.flink.walkthrough.common.table; - -import org.apache.flink.annotation.PublicEvolving; -import org.apache.flink.api.common.typeinfo.TypeInformation; -import org.apache.flink.api.common.typeinfo.Types; -import org.apache.flink.api.java.DataSet; -import org.apache.flink.api.java.operators.DataSink; -import org.apache.flink.streaming.api.datastream.DataStream; -import org.apache.flink.streaming.api.datastream.DataStreamSink; -import org.apache.flink.table.api.TableSchema; -import org.apache.flink.table.sinks.AppendStreamTableSink; -import org.apache.flink.table.sinks.BatchTableSink; -import org.apache.flink.table.sinks.TableSink; -import org.apache.flink.table.types.DataType; -import org.apache.flink.types.Row; -import org.apache.flink.walkthrough.common.sink.LoggerOutputFormat; - -/** - * A simple table sink for writing to stdout. - */ -@PublicEvolving -@SuppressWarnings("deprecation") -public class SpendReportTableSink implements AppendStreamTableSink<Row>, BatchTableSink<Row> { - - private final TableSchema schema; - - public SpendReportTableSink() { - this.schema = TableSchema - .builder() - .field("accountId", Types.LONG) - .field("timestamp", Types.SQL_TIMESTAMP) - .field("amount", Types.DOUBLE) - .build(); - } - - @Override - public DataSink<?> consumeDataSet(DataSet<Row> dataSet) { - return dataSet - .map(SpendReportTableSink::format) - .output(new LoggerOutputFormat()); - } - - @Override - public DataStreamSink<?> consumeDataStream(DataStream<Row> dataStream) { - return dataStream - .map(SpendReportTableSink::format) - .writeUsingOutputFormat(new LoggerOutputFormat()) - .setParallelism(dataStream.getParallelism()); - } - - @Override - public TableSchema getTableSchema() { - return schema; - } - - @Override - public DataType getConsumedDataType() { - return getTableSchema().toRowDataType(); - } - - @Override - public String[] getFieldNames() { - return getTableSchema().getFieldNames(); - } - - @Override - public TypeInformation<?>[] getFieldTypes() { - return getTableSchema().getFieldTypes(); - } - - @Override - public TableSink<Row> configure(String[] fieldNames, TypeInformation<?>[] fieldTypes) { - return this; - } - - private static String format(Row row) { - //noinspection MalformedFormatString - return String.format("%s, %s, $%.2f", row.getField(0), row.getField(1), row.getField(2)); - } - -} diff --git a/flink-walkthroughs/flink-walkthrough-common/src/main/java/org/apache/flink/walkthrough/common/table/TruncateDateToHour.java b/flink-walkthroughs/flink-walkthrough-common/src/main/java/org/apache/flink/walkthrough/common/table/TruncateDateToHour.java deleted file mode 100644 index d9f35b9..0000000 --- a/flink-walkthroughs/flink-walkthrough-common/src/main/java/org/apache/flink/walkthrough/common/table/TruncateDateToHour.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.flink.walkthrough.common.table; - -import org.apache.flink.annotation.PublicEvolving; -import org.apache.flink.api.common.typeinfo.TypeInformation; -import org.apache.flink.api.common.typeinfo.Types; -import org.apache.flink.table.functions.ScalarFunction; - -/** - * A user defined function for rounding timestamps down to - * the nearest hour. - */ -@PublicEvolving -@SuppressWarnings("unused") -public class TruncateDateToHour extends ScalarFunction { - - private static final long serialVersionUID = 1L; - - private static final long ONE_HOUR = 60 * 60 * 1000; - - public long eval(long timestamp) { - return timestamp - (timestamp % ONE_HOUR); - } - - @Override - public TypeInformation<?> getResultType(Class<?>[] signature) { - return Types.SQL_TIMESTAMP; - } -} diff --git a/flink-walkthroughs/flink-walkthrough-common/src/main/java/org/apache/flink/walkthrough/common/table/UnboundedTransactionTableSource.java b/flink-walkthroughs/flink-walkthrough-common/src/main/java/org/apache/flink/walkthrough/common/table/UnboundedTransactionTableSource.java deleted file mode 100644 index f6114da..0000000 --- a/flink-walkthroughs/flink-walkthrough-common/src/main/java/org/apache/flink/walkthrough/common/table/UnboundedTransactionTableSource.java +++ /dev/null @@ -1,89 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.flink.walkthrough.common.table; - -import org.apache.flink.annotation.PublicEvolving; -import org.apache.flink.api.common.functions.MapFunction; -import org.apache.flink.api.common.typeinfo.Types; -import org.apache.flink.streaming.api.datastream.DataStream; -import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; -import org.apache.flink.table.api.TableSchema; -import org.apache.flink.table.sources.DefinedRowtimeAttributes; -import org.apache.flink.table.sources.RowtimeAttributeDescriptor; -import org.apache.flink.table.sources.StreamTableSource; -import org.apache.flink.table.sources.tsextractors.ExistingField; -import org.apache.flink.table.sources.wmstrategies.BoundedOutOfOrderTimestamps; -import org.apache.flink.table.types.DataType; -import org.apache.flink.types.Row; -import org.apache.flink.walkthrough.common.entity.Transaction; -import org.apache.flink.walkthrough.common.source.TransactionSource; - -import java.sql.Timestamp; -import java.util.Collections; -import java.util.List; - -/** - * A table source for reading an unbounded set of transactions. - * - * <p>This table could be backed by a message queue or other streaming data source. - */ -@PublicEvolving -@SuppressWarnings({"deprecation", "unused"}) -public class UnboundedTransactionTableSource - implements StreamTableSource<Row>, - DefinedRowtimeAttributes { - - @Override - public DataStream<Row> getDataStream(StreamExecutionEnvironment execEnv) { - return execEnv - .addSource(new TransactionSource()) - .map(transactionRowMapFunction()) - .returns(getTableSchema().toRowType()); - } - - private MapFunction<Transaction, Row> transactionRowMapFunction() { - return transaction -> Row.of( - transaction.getAccountId(), - new Timestamp(transaction.getTimestamp()), - transaction.getAmount()); - } - - @Override - public DataType getProducedDataType() { - return getTableSchema().toRowDataType(); - } - - @Override - public TableSchema getTableSchema() { - return TableSchema.builder() - .field("accountId", Types.LONG) - .field("timestamp", Types.SQL_TIMESTAMP) - .field("amount", Types.DOUBLE) - .build(); - } - - @Override - public List<RowtimeAttributeDescriptor> getRowtimeAttributeDescriptors() { - return Collections.singletonList( - new RowtimeAttributeDescriptor( - "timestamp", - new ExistingField("timestamp"), - new BoundedOutOfOrderTimestamps(100))); - } -}
